Add anonymous web search and fetching to ReCall inference

README.md

@@ -110,6 +110,45 @@ bash train.sh \
 If you want to **fully reproduce** *ReCall*, please refer to the multi-node training script in `scripts/train/train_multi_node.sh`.
 
 ### Inference
+#### Web search without a sandbox
+
+With no executor or tool environment configured, the inference wrapper offers
+`web_search` and `web_fetch` through [Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp).
+Search uses Fast mode and needs no account or API key. A served ReCall model is
+still required. For an inference-only installation in a Python 3.10 environment:
+
+```bash
+pip install -r requirements-inference.txt
+pip install --no-deps -e .
+```
+
+```python
+from re_call import ReCall
+
+agent = ReCall(model_url="http://localhost:8000")
+response = agent.run(question="What are the latest releases of Python? Read the release notes and cite source URLs.")
+print(response)
+```
+
+The wrapper discovers the two tools over Streamable HTTP at
+`https://search.parallel.ai/mcp`, includes their schemas in the model prompt,
+and returns their results to the model in the existing tool loop. Each `run`
+starts a new conversation; related search and fetch calls, including generation
+retries, share one opaque identifier. The adapter sends no authentication and
+does not read API keys from the environment. The free service has usage limits;
+tool errors are returned to the model, and discovery errors stop the run.
+`search_timeout=30` sets the deadline in seconds for each discovery or tool request.
+
+Set `ReCall(model_url, search_enabled=False)` to disable the built-in web tools,
+or pass `func_schemas=[]` to expose no tools for one run. The API is synchronous;
+from async code, use `await asyncio.to_thread(agent.run, question=question)`.
+Supplying an executor URL, environment and schemas keeps the existing sandbox
+path below. Explicit sandbox configurations never fall back to web search.
+This integration applies to inference; the training and evaluation retrievers
+continue to use their configured environments.
+
+#### Custom sandbox tools
+
 This section demonstrates how to perform inference using the trained *ReCall* model. We provide a standard wrapper class in `src/re_call/inference/re_call.py` that simplifies the inference process. To get started, you only need to provide the model URL and sandbox URL, then use the `run` function to execute inference. The `ReCall` class handles all the orchestration between model generation and tool execution internally. For a practical example of using the `ReCall` class, please refer to our sample implementation at `scripts/inference/re_call_use_case.py`.
 
 For model serving, we recommend using [SGLang](https://docs.sglang.ai/). You can either download our open-source models or train your own models to conduct the inference. Here is an example of how to launch the model service:
@@ -175,4 +214,4 @@ If you find this work useful, please cite it as follows:
   primaryClass={cs.AI},
   url={https://arxiv.org/abs/2503.19470}, 
 }
-```
+```

requirements-inference.txt

@@ -0,0 +1,3 @@
+# The inference wrapper does not require the training or GPU dependencies.
+requests
+mcp>=1.26,<2

setup.py

@@ -54,6 +54,8 @@
   'jieba',
 
   # others
+  'requests',
+  'mcp>=1.26,<2',
   'sglang',
   'jsonlines',
 ]
@@ -73,4 +75,4 @@
     include_package_data=True,
     long_description=long_description,
     long_description_content_type='text/markdown'
-)
+)

src/re_call/inference/parallel_search.py

@@ -0,0 +1,95 @@
+"""Anonymous web tools for a single ReCall inference run."""
+
+import asyncio
+import copy
+import json
+import math
+from datetime import timedelta
+
+import anyio
+import httpx
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+
+ENDPOINT = "https://search.parallel.ai/mcp"
+USER_AGENT = "ReCall (+https://github.com/Agent-RL/ReCall)"
+TOOL_NAMES = frozenset({"web_search", "web_fetch"})
+
+
+class ParallelSearch:
+    """Stateless sync adapter; conversation identity is supplied by ReCall.run."""
+
+    def __init__(self, timeout=30):
+        if not math.isfinite(timeout) or timeout <= 0:
+            raise ValueError("search_timeout must be finite and positive")
+        self.timeout = timeout
+
+    async def _request(self, name=None, arguments=None):
+        # Bound discovery and tool execution, including transport setup/cleanup.
+        with anyio.fail_after(self.timeout):
+            async with httpx.AsyncClient(
+                headers={"User-Agent": USER_AGENT}, timeout=self.timeout,
+            ) as client:
+                async with streamable_http_client(ENDPOINT, http_client=client) as streams:
+                    async with ClientSession(
+                        streams[0], streams[1],
+                        read_timeout_seconds=timedelta(seconds=self.timeout),
+                    ) as session:
+                        await session.initialize()
+                        if name is None:
+                            return await session.list_tools()
+                        return await session.call_tool(name, arguments)
+
+    def schemas(self):
+        """Discover model-visible schemas without exposing host-owned metadata."""
+        tools = asyncio.run(self._request()).tools
+        schemas = []
+        for tool in tools:
+            if tool.name not in TOOL_NAMES:
+                continue
+            parameters = copy.deepcopy(tool.inputSchema)
+            for key in ("session_id", "model_name"):
+                parameters.get("properties", {}).pop(key, None)
+                if key in parameters.get("required", []):
+                    parameters["required"].remove(key)
+            schemas.append({
+                "name": tool.name, "description": tool.description or tool.name,
+                "parameters": parameters,
+            })
+        if {tool["name"] for tool in schemas} != TOOL_NAMES:
+            raise RuntimeError("Parallel Search MCP did not provide web_search and web_fetch")
+        return json.dumps(schemas)
+
+    def execute(self, tool_calls, session_id):
+        results = []
+        for raw_call in tool_calls:
+            try:
+                call = json.loads(raw_call)
+                name = call["name"]
+                arguments = call.get("arguments", {})
+                if name not in TOOL_NAMES or not isinstance(arguments, dict):
+                    raise ValueError("only web_search and web_fetch with object arguments are allowed")
+                arguments = dict(arguments)
+                # The model cannot replace the run's identity or invent its model ID.
+                arguments.pop("model_name", None)
+                arguments["session_id"] = session_id
+                result = asyncio.run(self._request(name, arguments))
+                text = "\n".join(
+                    block.text for block in result.content if block.type == "text"
+                )
+                if not text and result.structuredContent is not None:
+                    text = json.dumps(result.structuredContent)
+                results.append(("error: " if result.isError else "result: \n") + text)
+            except Exception as error:
+                # Cancellation/KeyboardInterrupt propagate; ordinary failures become
+                # observations, never a silent switch to a different provider.
+                results.append("error: Parallel Search MCP: " + self._error_text(error))
+        return results
+
+    @staticmethod
+    def _error_text(error):
+        children = getattr(error, "exceptions", None)
+        if children:
+            return "; ".join(ParallelSearch._error_text(child) for child in children)
+        return str(error) or type(error).__name__

src/re_call/inference/re_call.py

@@ -2,6 +2,7 @@
 import json
 import requests
 import time
+import uuid
 from typing import List
 from functools import wraps
 
@@ -40,9 +41,11 @@ class ReCall():
 {{"name": <function-name>, "arguments": <args-json-object>}}
 </tool_call>"""
 
-    def __init__(self, model_url, executor_url):
+    def __init__(self, model_url, executor_url=None, *, search_enabled=True, search_timeout=30):
         self.model_url = model_url
         self.executor_url = executor_url
+        self.search_enabled = search_enabled
+        self.search_timeout = search_timeout
 
     def init_prompt(self, func_schemas, question):
         system_prompt = f"<|im_start|>system\n{self.system_prompt.format(func_schemas=func_schemas)}<|im_end|>"
@@ -137,8 +140,37 @@ def extract_tool_calls(self, output_str):
         except Exception as e:
             return []
 
+    def run(self, env=None, func_schemas=None, question=None):
+        """Run inference with explicit sandbox tools or unconfigured web tools.
+
+        Each call starts an independent conversation. This is a synchronous API;
+        async applications should call it in a worker thread.
+        """
+        if question is None:
+            raise ValueError("question is required")
+        if env is None and func_schemas is None and self.executor_url is None:
+            if self.search_enabled:
+                from .parallel_search import ParallelSearch
+
+                search = ParallelSearch(timeout=self.search_timeout)
+                session_id = str(uuid.uuid4())
+                func_schemas = search.schemas()
+                execute = lambda calls: search.execute(calls, session_id)
+            else:
+                func_schemas = "[]"
+                execute = lambda calls: ["error: web tools are disabled"] * len(calls)
+        elif env is None and func_schemas in ([], "[]") and self.executor_url is None:
+            execute = lambda calls: ["error: no tools are configured"] * len(calls)
+        else:
+            # A partially configured sandbox is never an invitation to use the web.
+            if env is None or func_schemas is None or self.executor_url is None:
+                raise ValueError("sandbox tools require env, func_schemas and executor_url")
+            execute = lambda calls: self.execute_tool_calls(env, calls)
+        # Keep the ID and selected executor across the existing generation retry.
+        return self._run(func_schemas, question, execute)
+
     @retry(max=5, sleep=1)
-    def run(self, env, func_schemas, question):
+    def _run(self, func_schemas, question, execute):
         curr_prompt = self.init_prompt(func_schemas, question)
         for _ in range(5):
             response = requests.post(
@@ -157,7 +189,7 @@ def run(self, env, func_schemas, question):
             if len(tool_calls) == 0:
                 break
 
-            results: List[str] = self.execute_tool_calls(env, tool_calls)
+            results: List[str] = execute(tool_calls)
             curr_prompt = self.cat_tool_results(curr_prompt, tool_calls, results)
 
-        return curr_prompt
+        return curr_prompt