Add keyless Parallel search to unconfigured agent factories

README.md

@@ -103,6 +103,10 @@ chain = Chain(agent=agent, memory=memory)
 print(chain.run("Write me a poem about AI")['message'])
 ```
 
+Omitting `tools` enables keyless Parallel web search and page fetching. Pass
+`tools=[]` to disable tools, or supply your own list to keep full control of web
+access. See [tool setup and privacy](./docs/tool.md#parallel-web-search-and-page-fetching).
+
 Just like in LangChain, you can add a list of tools to the agent
 
 ```python

autochain/agent/conversational_agent/conversational_agent.py

@@ -19,6 +19,7 @@
 from autochain.agent.structs import AgentAction, AgentFinish
 from autochain.models.base import BaseLanguageModel, Generation
 from autochain.tools.base import Tool
+from autochain.tools.parallel_search import parallel_search_tools
 from autochain.utils import print_with_color
 
 logger = logging.getLogger(__name__)
@@ -51,7 +52,7 @@ def from_llm_and_tools(
         **kwargs: Any,
     ) -> ConversationalAgent:
         """Construct an agent from an LLM and tools."""
-        tools = tools or []
+        tools = parallel_search_tools() if tools is None else tools
 
         template = cls.get_prompt_template(
             template=prompt_template,

autochain/agent/openai_functions_agent/openai_functions_agent.py

@@ -13,6 +13,7 @@
 from autochain.agent.structs import AgentAction, AgentFinish
 from autochain.models.base import BaseLanguageModel, Generation
 from autochain.tools.base import Tool
+from autochain.tools.parallel_search import parallel_search_tools
 from autochain.utils import print_with_color
 from colorama import Fore
 
@@ -43,7 +44,7 @@ def from_llm_and_tools(
         min_confidence: int = 3,
         **kwargs: Any,
     ) -> OpenAIFunctionsAgent:
-        tools = tools or []
+        tools = parallel_search_tools() if tools is None else tools
 
         allowed_tools = {tool.name: tool for tool in tools}
         _output_parser = output_parser or OpenAIFunctionOutputParser()

autochain/chain/base_chain.py

@@ -13,6 +13,7 @@
 from autochain.chain import constants
 from autochain.memory.base import BaseMemory
 from autochain.tools.base import Tool
+from autochain.tools.parallel_search import search_conversation
 from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
@@ -90,7 +91,8 @@ def run(
         inputs = self.prep_inputs(user_query)
         logger.info(f"\n Input to agent: {inputs}")
         try:
-            output = self._run(inputs)
+            with search_conversation(self.memory):
+                output = self._run(inputs)
         except (KeyboardInterrupt, Exception) as e:
             raise e
 

autochain/memory/base.py

@@ -8,15 +8,27 @@
     Optional,
     Union,
 )
+from uuid import UUID, uuid4
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from autochain.agent.message import ChatMessageHistory, MessageType
 
 
 class BaseMemory(BaseModel, ABC):
     """Base interface for memory in chains."""
 
+    conversation_id: UUID = Field(default_factory=uuid4)
+    """Opaque conversation metadata, separate from retrievable user memories."""
+
+    def get_conversation_id(self) -> str:
+        """Return conversation identity without semantic retrieval."""
+        return str(self.conversation_id)
+
+    def reset_conversation_id(self) -> None:
+        """Start a fresh identity. Custom memory implementations call this in clear()."""
+        self.conversation_id = uuid4()
+
     @abstractmethod
     def load_memory(
         self, key: Union[str, None] = None, default: Optional[Any] = None, **kwargs: Any

autochain/memory/buffer_memory.py

@@ -36,5 +36,6 @@ def save_conversation(
 
     def clear(self) -> None:
         """Clear memory contents."""
+        self.reset_conversation_id()
         self.conversation_history.clear()
         self.kv_memory = {}

autochain/memory/long_term_memory.py

@@ -66,6 +66,7 @@ def save_conversation(
 
     def clear(self) -> None:
         """Clear memory contents."""
+        self.reset_conversation_id()
         self.conversation_history.clear()
         self.long_term_memory.clear_index()
         self.kv_memory = {}

autochain/memory/redis_memory.py

@@ -1,5 +1,6 @@
 import pickle
 from typing import Any, Optional, Dict
+from uuid import UUID
 
 from autochain.agent.message import (
     ChatMessageHistory,
@@ -51,6 +52,21 @@ def save_memory(self, key: str, value: Any) -> None:
         pickled = pickle.dumps(value)
         self.redis_client.set(key, pickled, ex=self.expire_time)
 
+    def get_conversation_id(self) -> str:
+        """Persist typed metadata independently of pickled conversation contents."""
+        key = self.redis_key_prefix + ":_autochain_conversation_id"
+        saved = self.redis_client.get(key)
+        try:
+            if isinstance(saved, bytes):
+                saved = saved.decode("ascii")
+            if not isinstance(saved, str):
+                raise ValueError("Missing conversation identity")
+            self.conversation_id = UUID(saved)
+        except (ValueError, UnicodeError):
+            pass
+        self.redis_client.set(key, str(self.conversation_id), ex=self.expire_time)
+        return str(self.conversation_id)
+
     def save_conversation(
         self, message: str, message_type: MessageType, **kwargs
     ) -> None:
@@ -77,3 +93,4 @@ def clear(self) -> None:
         """Clear redis memory."""
         for key in self.redis_client.keys(f"{self.redis_key_prefix}:*"):
             self.redis_client.delete(key)
+        self.reset_conversation_id()

autochain/tools/base.py

@@ -102,7 +102,7 @@ def run(
         try:
             tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
             tool_output = self._run(*tool_args, **tool_kwargs)
-        except (Exception, KeyboardInterrupt) as e:
+        except Exception as e:
             raise ToolRunningError(
                 message=f"Failed to run tool {self.name} due to {e}"
             ) from e

autochain/tools/parallel_search.py

@@ -0,0 +1,194 @@
+"""Keyless search and page extraction through Parallel's HTTP MCP endpoint."""
+import json
+import socket
+from contextlib import contextmanager
+from contextvars import ContextVar
+from typing import List
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlsplit
+from urllib.request import Request, urlopen
+from uuid import uuid4
+
+from autochain.tools.base import Tool
+
+ENDPOINT = "https://search.parallel.ai/mcp"
+USER_AGENT = "AutoChain (https://github.com/Forethought-Technologies/AutoChain)"
+_conversation = ContextVar("autochain_parallel_conversation", default=None)
+_MAX_RESPONSE = 2 * 1024 * 1024
+
+
+@contextmanager
+def search_conversation(memory):
+    """Bind search identity to chain memory; clearing memory starts a new chat."""
+    session_id = memory.get_conversation_id() if memory is not None else str(uuid4())
+    token = _conversation.set(session_id)
+    try:
+        yield
+    finally:
+        _conversation.reset(token)
+
+
+class _Client:
+    """Small synchronous Streamable HTTP client for this fixed server only."""
+
+    def __init__(self, timeout):
+        self.timeout = timeout
+        self.session = None
+
+    def request(self, method, params, request_id=None):
+        message = {"jsonrpc": "2.0", "method": method, "params": params}
+        if request_id is not None:
+            message["id"] = request_id
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json, text/event-stream",
+            "User-Agent": USER_AGENT,
+        }
+        if self.session:
+            headers["Mcp-Session-Id"] = self.session
+            headers["MCP-Protocol-Version"] = "2025-03-26"
+        request = Request(ENDPOINT, data=json.dumps(message).encode(), headers=headers)
+        with urlopen(request, timeout=self.timeout) as response:
+            if method == "initialize":
+                self.session = response.headers.get("Mcp-Session-Id")
+            if request_id is None:
+                return None
+            if "text/event-stream" in response.headers.get("Content-Type", ""):
+                # Stop at the matching event even if the server keeps SSE open.
+                size, data = 0, []
+                result = None
+                while size <= _MAX_RESPONSE:
+                    line = response.readline(_MAX_RESPONSE - size + 1)
+                    size += len(line)
+                    if size > _MAX_RESPONSE:
+                        raise ValueError("Parallel response exceeds 2 MiB")
+                    if not line:
+                        break
+                    text = line.decode().rstrip("\r\n")
+                    if text.startswith("data:"):
+                        data.append(text[5:].lstrip(" "))
+                    elif not text and data:
+                        message = json.loads("\n".join(data))
+                        data = []
+                        if (
+                            isinstance(message, dict)
+                            and message.get("id") == request_id
+                        ):
+                            result = message
+                            break
+            else:
+                body = response.read(_MAX_RESPONSE + 1)
+                if len(body) > _MAX_RESPONSE:
+                    raise ValueError("Parallel response exceeds 2 MiB")
+                result = json.loads(body)
+            if (
+                not isinstance(result, dict)
+                or result.get("id") != request_id
+                or result.get("jsonrpc") != "2.0"
+            ):
+                raise ValueError("Invalid Parallel MCP response")
+            if "error" in result:
+                raise ValueError("Parallel MCP returned an error")
+            if not isinstance(result.get("result"), dict):
+                raise ValueError("Invalid Parallel MCP result")
+            return result["result"]
+
+    def call(self, name, arguments):
+        try:
+            initialized = self.request(
+                "initialize",
+                {
+                    "protocolVersion": "2025-03-26",
+                    "capabilities": {},
+                    "clientInfo": {"name": "autochain", "version": "0.0.5"},
+                },
+                1,
+            )
+            if initialized.get("protocolVersion") != "2025-03-26":
+                raise ValueError("Unsupported Parallel MCP protocol version")
+            self.request("notifications/initialized", {})
+            tools = self.request("tools/list", {}, 2)["tools"]
+            if name not in [tool["name"] for tool in tools]:
+                raise ValueError("Parallel MCP tool is unavailable")
+            # Conversation identity is separate from the MCP transport session.
+            arguments["session_id"] = _conversation.get() or str(uuid4())
+            result = self.request(
+                "tools/call", {"name": name, "arguments": arguments}, 3
+            )
+            text = "\n".join(
+                item["text"]
+                for item in result.get("content", [])
+                if item["type"] == "text"
+            )
+            if result.get("isError"):
+                return "Parallel tool error: " + text
+            if not text:
+                raise ValueError("Parallel returned no text")
+            return text
+        except HTTPError as error:
+            return f"Parallel unavailable (HTTP {error.code}); retry later or use another configured tool."
+        except (URLError, TimeoutError, socket.timeout):
+            return "Parallel request failed or timed out; retry later."
+        except (ValueError, KeyError, TypeError):
+            return "Parallel returned an invalid MCP response; retry later."
+        finally:
+            if self.session:
+                request = Request(
+                    ENDPOINT,
+                    method="DELETE",
+                    headers={
+                        "Mcp-Session-Id": self.session,
+                        "MCP-Protocol-Version": "2025-03-26",
+                        "User-Agent": USER_AGENT,
+                    },
+                )
+                try:
+                    with urlopen(request, timeout=self.timeout):
+                        pass
+                except (OSError, ValueError):
+                    pass  # Session cleanup must not hide the tool's result.
+
+
+def parallel_search_tools(timeout: float = 30.0) -> List[Tool]:
+    """Create search/fetch tools without connecting or reading ambient API keys.
+
+    The timeout is in seconds per HTTP request. Standalone calls each start a
+    research run; calls dispatched by a Chain share its conversation identity.
+    """
+    if timeout <= 0:
+        raise ValueError("timeout must be positive")
+
+    # Plain functions keep compatibility with ChatOpenAI's signature inspection.
+    def web_search(query: str):
+        if not query.strip():
+            return "Provide a nonempty search query."
+        return _Client(timeout).call(
+            "web_search", {"objective": query, "search_queries": [query]}
+        )
+
+    def web_fetch(url: str):
+        try:
+            parsed = urlsplit(url)
+        except ValueError:
+            return "Provide an HTTP or HTTPS URL without credentials."
+        if (
+            parsed.scheme not in ("http", "https")
+            or not parsed.hostname
+            or parsed.username
+            or parsed.password
+        ):
+            return "Provide an HTTP or HTTPS URL without credentials."
+        return _Client(timeout).call("web_fetch", {"urls": [url]})
+
+    return [
+        Tool(
+            name="web_search",
+            description="Search the web for current information. Input: a search query. Returns source URLs and excerpts. Powered by Parallel Search MCP.",
+            func=web_search,
+        ),
+        Tool(
+            name="web_fetch",
+            description="Read a web page for evidence. Input: an HTTP or HTTPS URL, usually from web_search. Returns page excerpts.",
+            func=web_fetch,
+        ),
+    ]

docs/tool.md

@@ -31,6 +31,44 @@ formatted into the prompt when using `OpenAIFunctionsAgent`.
 
 
 ## Tools included
+### Parallel web search and page fetching
+
+`ConversationalAgent.from_llm_and_tools(llm=llm)` and
+`OpenAIFunctionsAgent.from_llm_and_tools(llm=llm)` now include `web_search` and
+`web_fetch` when `tools` is omitted or `None`. These tools use
+[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp)
+at `https://search.parallel.ai/mcp`. Search runs in Fast mode with no account or
+API key required. Anonymous use has service limits; failures are returned to the
+agent as tool output without switching providers.
+
+The existing `ChatOpenAI` adapter uses OpenAI's pre-1.0 Python API. Use the
+version recorded in `poetry.lock` (`openai==0.27.8`) for that adapter; the web
+tools themselves do not require OpenAI or Parallel credentials.
+
+Pass `tools=[]` to disable all tools and keep the agent offline apart from model
+inference. Any explicit tool list remains authoritative, including Google search
+or custom tools. Constructing an agent does not make network requests. Calling
+these tools sends the query or URL to Parallel; only enable them when web access
+is appropriate for your application.
+
+To combine them with your own tools or adjust the per-request timeout:
+
+```python
+from autochain.tools.parallel_search import parallel_search_tools
+
+tools = parallel_search_tools(timeout=20) + your_other_tools
+agent = OpenAIFunctionsAgent.from_llm_and_tools(llm=llm, tools=tools)
+```
+
+The synchronous client uses Python's standard library, including on Python 3.8.
+Each HTTP request has a 30-second timeout by default. A chain keeps one opaque
+search conversation identifier in its memory across turns; clear the memory to
+start a new conversation. Separate chains should use separate memory instances.
+Standalone `Tool.run` calls each start a separate research run. Ctrl-C cancels
+execution rather than converting the interruption to a tool error.
+Custom memory implementations should keep `get_conversation_id()` separate from
+semantic retrieval and call `reset_conversation_id()` in their `clear()` method.
+
 ### GoogleSearchTool
 Migrated from LangChain, which is also an example for user to easily migrate any tool from 
 LangChain if needed.  
@@ -52,4 +90,4 @@ long term memory for the agent.
 ### LanceDBTool
 Internal search tool that can be used for long term memory of the agent or looking up relevant
 information that does not exists from the Internet. Currently, AutoChain supports `ChromaDB` as
-long term memory for the agent. LanceDBTool is serverless, and does not require any setup.
+long term memory for the agent. LanceDBTool is serverless, and does not require any setup.