From 6fdc598169b1e97c4c822a2bd95965d5c3360c39 Mon Sep 17 00:00:00 2001 From: Parallel Date: Tue, 20 Oct 2026 13:52:34 +0000 Subject: [PATCH 1/3] Add keyless Parallel search for research projects --- agentstack/_tools/parallel/__init__.py | 89 +++++++++ agentstack/_tools/parallel/config.json | 8 + agentstack/cli/run.py | 20 +- agentstack/search.py | 28 +++ agentstack/templates/__init__.py | 14 +- docs/tools/community.mdx | 1 + docs/tools/tool/parallel.mdx | 53 +++++ pyproject.toml | 1 + tests/test_research_search_default.py | 41 ++++ tests/tools/test_tool_parallel.py | 262 +++++++++++++++++++++++++ 10 files changed, 507 insertions(+), 10 deletions(-) create mode 100644 agentstack/_tools/parallel/__init__.py create mode 100644 agentstack/_tools/parallel/config.json create mode 100644 agentstack/search.py create mode 100644 docs/tools/tool/parallel.mdx create mode 100644 tests/test_research_search_default.py create mode 100644 tests/tools/test_tool_parallel.py diff --git a/agentstack/_tools/parallel/__init__.py b/agentstack/_tools/parallel/__init__.py new file mode 100644 index 00000000..8253941d --- /dev/null +++ b/agentstack/_tools/parallel/__init__.py @@ -0,0 +1,89 @@ +"""Free anonymous search and page excerpts through Parallel Search MCP.""" + +import json +from datetime import timedelta +from urllib.parse import urlsplit + +from agentstack.search import get_search_session_id +from agentstack.utils import get_version + + +MCP_URL = 'https://search.parallel.ai/mcp' +REQUEST_TIMEOUT = timedelta(seconds=30) + + +async def _call_tool(name: str, arguments: dict) -> str: + # Imported only when called, so listing tools does not require the MCP extra. + from mcp import ClientSession + from mcp.client.streamable_http import streamablehttp_client + + headers = {'User-Agent': f'AgentStack/{get_version()}'} + result = None + async with streamablehttp_client( + MCP_URL, + headers=headers, + timeout=REQUEST_TIMEOUT, + sse_read_timeout=REQUEST_TIMEOUT, + ) as (read, write, _): + async with ClientSession(read, write, read_timeout_seconds=REQUEST_TIMEOUT) as session: + await session.initialize() + result = await session.call_tool(name, arguments, read_timeout_seconds=REQUEST_TIMEOUT) + + if result is None: + # A transport task can be cancelled while its task group unwinds. + # Preserve cancellation even if an SDK context manager suppresses it. + from anyio import get_cancelled_exc_class + + raise get_cancelled_exc_class() + text = '\n'.join(block.text for block in result.content if block.type == 'text') + if result.isError: + raise RuntimeError(f'Parallel {name} failed: {text}') + if text: + return text + if result.structuredContent is not None: + return json.dumps(result.structuredContent) + raise RuntimeError(f'Parallel {name} returned no text or structured content') + + +def _run_tool(name: str, arguments: dict) -> str: + from anyio.from_thread import start_blocking_portal + + arguments['session_id'] = get_search_session_id() + # Framework tools are synchronous. A portal also works when their caller + # already runs an event loop, without nesting asyncio.run in that loop. + with start_blocking_portal() as portal: + return portal.call(_call_tool, name, arguments) + + +def web_search(objective: str, search_queries: list[str]) -> str: + """Search the web for current information, returning source URLs and excerpts. + + Args: + objective: A focused description of the information to find. + search_queries: At least one related keyword query; prefer 3–6 words each. + + Returns: + Search results with URLs and excerpts to cite or read with web_fetch. + """ + if not objective.strip() or not search_queries or any(not query.strip() for query in search_queries): + raise ValueError('Provide an objective and at least one nonempty search query') + return _run_tool('web_search', {'objective': objective, 'search_queries': search_queries}) + + +def web_fetch(urls: list[str], objective: str = '') -> str: + """Read excerpts from up to 20 HTTP(S) pages, preserving source URLs. + + Args: + urls: The web page URLs to read, such as sources returned by web_search. + objective: Optional description of relevant information, up to 200 characters. + + Returns: + Page excerpts and their source URLs for the agent to analyze and cite. + """ + if not 1 <= len(urls) <= 20: + raise ValueError('Provide between 1 and 20 URLs') + if any(urlsplit(url).scheme not in ('http', 'https') or not urlsplit(url).netloc for url in urls): + raise ValueError('Only absolute HTTP or HTTPS URLs are supported') + if len(objective) > 200: + raise ValueError('The fetch objective must be at most 200 characters') + return _run_tool('web_fetch', {'urls': urls, 'objective': objective or None}) diff --git a/agentstack/_tools/parallel/config.json b/agentstack/_tools/parallel/config.json new file mode 100644 index 00000000..5172262b --- /dev/null +++ b/agentstack/_tools/parallel/config.json @@ -0,0 +1,8 @@ +{ + "name": "parallel", + "url": "https://docs.parallel.ai/integrations/mcp/search-mcp", + "category": "web-retrieval", + "dependencies": ["mcp>=1.12,<2"], + "tools": ["web_search", "web_fetch"], + "cta": "Parallel Search MCP is ready. No search API key or account is required." +} diff --git a/agentstack/cli/run.py b/agentstack/cli/run.py index 3e142b19..5853cb2c 100644 --- a/agentstack/cli/run.py +++ b/agentstack/cli/run.py @@ -10,6 +10,7 @@ from agentstack.exceptions import ValidationError from agentstack import inputs from agentstack import frameworks +from agentstack.search import search_session from agentstack.utils import get_framework, verify_agentstack_project MAIN_FILENAME: Path = Path("src/main.py") @@ -98,7 +99,7 @@ def run_project(command: str = 'run', cli_args: Optional[List[str]] = None): """Validate that the project is ready to run and then run it.""" conf.assert_project() verify_agentstack_project() - + if conf.get_framework() not in frameworks.SUPPORTED_FRAMEWORKS: raise ValidationError(f"Framework {conf.get_framework()} is not supported by agentstack.") @@ -122,14 +123,15 @@ def run_project(command: str = 'run', cli_args: Optional[List[str]] = None): # import src/main.py from the project path and run `command` from the project's main.py try: log.notify("Running your agent...") - project_main = _import_project_module(conf.PATH) - main = getattr(project_main, command) - - # handle both async and sync entrypoints - if asyncio.iscoroutinefunction(main): - asyncio.run(main()) - else: - main() + with search_session(): + project_main = _import_project_module(conf.PATH) + main = getattr(project_main, command) + + # handle both async and sync entrypoints + if asyncio.iscoroutinefunction(main): + asyncio.run(main()) + else: + main() except ImportError as e: raise ValidationError(f"Failed to import AgentStack project at: {conf.PATH.absolute()}\n{e}") except Exception as e: diff --git a/agentstack/search.py b/agentstack/search.py new file mode 100644 index 00000000..cb4a11fd --- /dev/null +++ b/agentstack/search.py @@ -0,0 +1,28 @@ +"""Conversation context shared by web search and page fetching.""" + +from contextlib import contextmanager +from contextvars import ContextVar +from uuid import uuid4 + + +_session_id: ContextVar[str | None] = ContextVar('agentstack_search_session', default=None) + + +@contextmanager +def search_session(): + """Start a research run; related tool calls share an opaque conversation ID. + + The CLI starts a session for each ``agentstack run``. Applications using the + tools directly can wrap their conversation in this context manager instead. + Nested sessions represent new conversations and restore the outer context. + """ + token = _session_id.set(str(uuid4())) + try: + yield + finally: + _session_id.reset(token) + + +def get_search_session_id() -> str: + """Use the current conversation, or a new ID for a standalone tool call.""" + return _session_id.get() or str(uuid4()) diff --git a/agentstack/templates/__init__.py b/agentstack/templates/__init__.py index 1602fc56..d1b3c3ff 100644 --- a/agentstack/templates/__init__.py +++ b/agentstack/templates/__init__.py @@ -4,6 +4,7 @@ import pydantic import requests import json +from dotenv import dotenv_values from agentstack.exceptions import ValidationError from agentstack.utils import get_package_path @@ -247,7 +248,18 @@ def from_template_name(cls, name: str) -> 'TemplateConfig': raise ValidationError(f"Template {name} not bundled with agentstack.") path = _get_builtin_templates_path() / f"{name}.json" - return cls.from_file(path) + template = cls.from_file(path) + # Only choose a default for new projects using the bundled research + # starter. Saved projects and explicit file/URL templates keep their tools. + if name == 'research': + home_key = dotenv_values(Path.home() / '.env').get('PERPLEXITY_API_KEY') + key = os.getenv('PERPLEXITY_API_KEY', home_key) + if not key or not key.strip(): + template.tools = [ + cls.Tool(name='parallel', agents=tool.agents) if tool.name == 'perplexity' else tool + for tool in template.tools + ] + return template @classmethod def from_file(cls, path: Path) -> 'TemplateConfig': diff --git a/docs/tools/community.mdx b/docs/tools/community.mdx index e70849e2..4f807afe 100644 --- a/docs/tools/community.mdx +++ b/docs/tools/community.mdx @@ -16,6 +16,7 @@ description: 'AgentStack tools from community contributors' - [Hyperbrowser](/tools/tool/hyperbrowser) ## Search +- [Parallel](/tools/tool/parallel) - [Perplexity](/tools/tool/perplexity) - [Dappier](/tools/tool/dappier) diff --git a/docs/tools/tool/parallel.mdx b/docs/tools/tool/parallel.mdx new file mode 100644 index 00000000..e17ac60d --- /dev/null +++ b/docs/tools/tool/parallel.mdx @@ -0,0 +1,53 @@ +--- +title: 'Parallel' +description: 'Free web search and page excerpts without a search API key' +--- + +[Parallel Search MCP](https://docs.parallel.ai/integrations/mcp/search-mcp) +provides web search powered by Fast mode and page fetching. No Parallel account +or API key is needed for light use; the anonymous service has usage limits. + +## Installation + +Within an AgentStack project, run: + +```bash +agentstack tools add parallel +``` + +This installs the MCP client and adds `web_search` and `web_fetch` to your agents. +Use `--agent researcher` to restrict the tool to one agent. The tools return source +URLs and excerpts for the agent to analyze and cite. `web_fetch` accepts up to 20 +HTTP(S) URLs and an optional objective of up to 200 characters. Requests time out +after 30 seconds; service errors propagate to the framework without switching +providers or retrying automatically. + +## Research starter + +New projects created with `agentstack init my_research --template research` use +Parallel when `PERPLEXITY_API_KEY` is absent from the environment and `~/.env`. +If that key is configured, the starter keeps Perplexity. An LLM provider must +still be configured for the agents. + +Existing projects and templates loaded from a file or URL keep their selected +tools. You can explicitly add Perplexity with `agentstack tools add perplexity`. +To remove Parallel from an existing project, run `agentstack tools remove parallel`. + +## Conversation context + +Each `agentstack run` gives related search and fetch calls one opaque session ID. +A later run starts a new session. Applications calling the tools without the CLI +can scope related calls with `search_session`: + +```python +import agentstack +from agentstack.search import search_session + +with search_session(): + search, fetch = agentstack.tools['parallel'] + # Pass these framework-native callables to your agent within this context. +``` + +Without a session context, each standalone tool call receives its own session ID. +Concurrent conversations have separate contexts. The MCP connection is opened +only when a tool is called; adding or listing the tool makes no search request. diff --git a/pyproject.toml b/pyproject.toml index aa045f76..04fe9690 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ dev = [ ] test = [ "tox", + "mcp>=1.12,<2", ] crewai = [ "crewai==0.118.0", diff --git a/tests/test_research_search_default.py b/tests/test_research_search_default.py new file mode 100644 index 00000000..0c5018a5 --- /dev/null +++ b/tests/test_research_search_default.py @@ -0,0 +1,41 @@ +import json +import os +from pathlib import Path +from unittest.mock import patch + +from agentstack.templates import TemplateConfig + + +def test_unconfigured_research_uses_parallel(tmp_path): + with patch.dict(os.environ, {}, clear=True), patch.object(Path, 'home', return_value=tmp_path): + template = TemplateConfig.from_template_name('research') + assert [(tool.name, tool.agents) for tool in template.tools] == [('parallel', ['researcher'])] + + +def test_configured_perplexity_is_preserved(tmp_path): + for environment, home in [ + ({'PERPLEXITY_API_KEY': 'configured'}, ''), + ({}, 'PERPLEXITY_API_KEY=home-key'), + ]: + (tmp_path / '.env').write_text(home) + with ( + patch.dict(os.environ, environment, clear=True), + patch.object(Path, 'home', return_value=tmp_path), + ): + assert TemplateConfig.from_template_name('research').tools[0].name == 'perplexity' + + +def test_explicit_file_template_keeps_selected_provider(tmp_path): + data = json.loads((Path(__file__).parent.parent / 'agentstack/templates/research.json').read_text()) + template_path = tmp_path / 'research.json' + template_path.write_text(json.dumps(data)) + with patch.dict(os.environ, {}, clear=True), patch.object(Path, 'home', return_value=tmp_path): + assert TemplateConfig.from_user_input(str(template_path)).tools[0].name == 'perplexity' + data['tools'] = [] + template_path.write_text(json.dumps(data)) + assert TemplateConfig.from_user_input(str(template_path)).tools == [] + + +def test_other_starters_keep_their_tools(tmp_path): + with patch.dict(os.environ, {}, clear=True), patch.object(Path, 'home', return_value=tmp_path): + assert TemplateConfig.from_template_name('hello_alex').tools[0].name == 'file_read' diff --git a/tests/tools/test_tool_parallel.py b/tests/tools/test_tool_parallel.py new file mode 100644 index 00000000..14994520 --- /dev/null +++ b/tests/tools/test_tool_parallel.py @@ -0,0 +1,262 @@ +"""Scripted runtime dispatch with real MCP protocol handling and mocked HTTP.""" + +import asyncio +import json +import shutil +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path + +import httpx +import pytest + +import agentstack +from agentstack import conf +from agentstack.cli.run import run_project +from agentstack.search import search_session +from agentstack.utils import get_version + + +pytest.importorskip('mcp') +AIMessage = pytest.importorskip('langchain_core.messages').AIMessage +graph_module = pytest.importorskip('langgraph.graph') +StateGraph, MessagesState = graph_module.StateGraph, graph_module.MessagesState +START, END = graph_module.START, graph_module.END +ToolNode = pytest.importorskip('langgraph.prebuilt').ToolNode + + +@pytest.fixture +def project(tmp_path, monkeypatch): + old_path = conf.PATH + monkeypatch.setenv('AGENTSTACK_TELEMETRY_OPT_OUT', '1') + monkeypatch.delenv('PERPLEXITY_API_KEY', raising=False) + monkeypatch.delenv('PARALLEL_API_KEY', raising=False) + monkeypatch.setattr(Path, 'home', lambda: tmp_path / 'home') + (tmp_path / 'agentstack.json').write_text(json.dumps({'framework': 'langgraph', 'tools': ['parallel']})) + (tmp_path / 'src').mkdir() + shutil.copy( + Path(__file__).parent.parent / 'fixtures/frameworks/langgraph/entrypoint_max.py', + tmp_path / 'src/graph.py', + ) + conf.set_path(tmp_path) + yield tmp_path + conf.set_path(old_path) + + +@pytest.fixture +def server(monkeypatch): + requests = [] + behavior = {'error': None, 'tool_error': False, 'structured': False} + original_client = httpx.AsyncClient + + def handler(request): + assert str(request.url) == 'https://search.parallel.ai/mcp' + assert request.headers['user-agent'] == f'AgentStack/{get_version()}' + assert 'authorization' not in request.headers + assert request.extensions['timeout']['read'] == 30 + if request.method != 'POST': + return httpx.Response(405) + body = json.loads(request.content) + requests.append(body) + method = body['method'] + if method == 'notifications/initialized': + return httpx.Response(202) + if method == 'initialize': + result = { + 'protocolVersion': body['params']['protocolVersion'], + 'capabilities': {'tools': {}}, + 'serverInfo': {'name': 'parallel-fixture', 'version': '1'}, + } + elif method == 'tools/list': + result = { + 'tools': [ + {'name': name, 'inputSchema': {'type': 'object', 'properties': {}}} + for name in ['web_search', 'web_fetch'] + ] + } + else: + assert method == 'tools/call' + if behavior['error']: + raise behavior['error'] + text = ( + 'Quota exceeded' + if behavior['tool_error'] + else 'Source: https://example.com/research\nPage excerpt' + ) + result = {'content': [{'type': 'text', 'text': text}], 'isError': behavior['tool_error']} + if behavior['structured']: + result = {'content': [], 'structuredContent': {'url': 'https://example.com/research'}} + return httpx.Response(200, json={'jsonrpc': '2.0', 'id': body['id'], 'result': result}) + + def client(*args, **kwargs): + kwargs['transport'] = httpx.MockTransport(handler) + return original_client(*args, **kwargs) + + monkeypatch.setattr(httpx, 'AsyncClient', client) + yield requests, behavior + + +def dispatcher(): + graph = StateGraph(MessagesState) + graph.add_node('tools', ToolNode(agentstack.tools['parallel'], handle_tool_errors=False)) + # Scripted consumer stands in for the next model turn, retaining source URLs. + graph.add_node('consume', lambda state: {'messages': [AIMessage(content=state['messages'][-1].content)]}) + graph.add_edge(START, 'tools') + graph.add_edge('tools', 'consume') + graph.add_edge('consume', END) + return graph.compile() + + +def invoke(app, name='web_search', arguments=None): + arguments = ( + arguments + if arguments is not None + else {'objective': 'Find research', 'search_queries': ['research sources']} + ) + message = AIMessage( + content='', tool_calls=[{'name': name, 'args': arguments, 'id': 'call-1', 'type': 'tool_call'}] + ) + return app.invoke({'messages': [message]})['messages'][-1].content + + +def calls(server): + return [request['params'] for request in server[0] if request['method'] == 'tools/call'] + + +def test_search_fetch_and_followup_reach_consumer(project, server): + app = dispatcher() + with search_session(): + assert 'https://example.com/research' in invoke(app) + assert 'Page excerpt' in invoke(app, 'web_fetch', {'urls': ['https://example.com/research']}) + invoke(app) + assert [call['name'] for call in calls(server)] == ['web_search', 'web_fetch', 'web_search'] + assert len({call['arguments']['session_id'] for call in calls(server)}) == 1 + assert calls(server)[1]['arguments']['urls'] == ['https://example.com/research'] + assert calls(server)[0]['arguments']['search_queries'] == ['research sources'] + + +def test_new_and_concurrent_conversations_use_distinct_ids(project, server): + app = dispatcher() + + def conversation(): + with search_session(): + invoke(app) + invoke(app, 'web_fetch', {'urls': ['https://example.com/research']}) + + with ThreadPoolExecutor(max_workers=2) as executor: + list(executor.map(lambda _: conversation(), range(2))) + counts = {} + for call in calls(server): + identifier = call['arguments']['session_id'] + counts[identifier] = counts.get(identifier, 0) + 1 + assert sorted(counts.values()) == [2, 2] + conversation() + assert calls(server)[-1]['arguments']['session_id'] not in counts + + +def test_cli_starts_new_session_and_reuses_it_within_run(project, server): + # Import the same real dispatcher used above from an actual project entrypoint. + module = __name__ + (project / 'src/main.py').write_text( + f'from {module} import dispatcher, invoke\n' + 'def run():\n' + ' app = dispatcher()\n' + ' invoke(app)\n' + ' invoke(app, "web_fetch", {"urls": ["https://example.com/research"]})\n' + ) + run_project() + run_project() + ids = [call['arguments']['session_id'] for call in calls(server)] + assert ids[0] == ids[1] + assert ids[2] == ids[3] + assert ids[0] != ids[2] + + +def test_async_cli_entrypoint_can_call_sync_tools(project, server): + (project / 'src/main.py').write_text( + f'from {__name__} import dispatcher, invoke\n' + 'async def run():\n' + ' invoke(dispatcher())\n' + ' invoke(dispatcher())\n' + ) + run_project() + assert len(calls(server)) == 2 + assert calls(server)[0]['arguments']['session_id'] == calls(server)[1]['arguments']['session_id'] + + +def test_explicit_perplexity_error_does_not_fall_back(project, server, monkeypatch): + import requests + + (project / 'agentstack.json').write_text(json.dumps({'framework': 'langgraph', 'tools': ['perplexity']})) + destinations = [] + + def fail(method, url, **kwargs): + destinations.append(url) + return type('Response', (), {'status_code': 503, 'text': 'Service unavailable'})() + + monkeypatch.setattr(requests, 'request', fail) + query = agentstack.tools['perplexity'][0] + assert query('research sources') == 'Failed to query perplexity' + assert destinations == ['https://api.perplexity.ai/chat/completions'] + assert conf.ConfigFile().tools == ['perplexity'] + assert server[0] == [] + + +def test_quota_error_and_retry_keep_conversation(project, server): + app = dispatcher() + with search_session(): + server[1]['tool_error'] = True + with pytest.raises(RuntimeError, match='Quota exceeded'): + invoke(app) + server[1]['tool_error'] = False + assert 'Page excerpt' in invoke(app) + assert calls(server)[0]['arguments']['session_id'] == calls(server)[1]['arguments']['session_id'] + + +@pytest.mark.parametrize('error', [httpx.ReadTimeout('timed out'), asyncio.CancelledError()]) +def test_timeout_and_cancellation_do_not_return_success(project, server, error): + server[1]['error'] = error + with pytest.raises(BaseException) as caught: + invoke(dispatcher()) + + def leaves(exception): + if hasattr(exception, 'exceptions'): + return [leaf for nested in exception.exceptions for leaf in leaves(nested)] + return [exception] + + # The synchronous portal represents async cancellation as Future cancellation. + from concurrent.futures import CancelledError + + expected = httpx.ReadTimeout if isinstance(error, httpx.ReadTimeout) else CancelledError + assert any(isinstance(leaf, expected) for leaf in leaves(caught.value)), repr(caught.value) + assert len(calls(server)) == 1 # No hidden retry, provider fallback or new identity. + + +def test_structured_content_is_consumed(project, server): + server[1]['structured'] = True + assert json.loads(invoke(dispatcher())) == {'url': 'https://example.com/research'} + + +@pytest.mark.parametrize( + 'name,arguments', + [ + ('web_search', {'objective': '', 'search_queries': ['research']}), + ('web_search', {'objective': 'research', 'search_queries': []}), + ('web_fetch', {'urls': ['file:///etc/passwd']}), + ('web_fetch', {'urls': []}), + ('web_fetch', {'urls': ['https://example.com'], 'objective': 'x' * 201}), + ], +) +def test_invalid_input_never_contacts_server(project, server, name, arguments): + with pytest.raises(ValueError): + invoke(dispatcher(), name, arguments) + assert server[0] == [] + + +def test_removed_tool_is_not_available_to_dispatcher(project, server): + graph = StateGraph(MessagesState) + graph.add_node('tools', ToolNode([])) + graph.add_edge(START, 'tools') + graph.add_edge('tools', END) + result = invoke(graph.compile()) + assert 'not a valid tool' in result + assert server[0] == [] From c664fae21ec5711a520a6bd0796eef58627f9fe9 Mon Sep 17 00:00:00 2001 From: Parallel Date: Tue, 20 Oct 2026 13:53:11 +0000 Subject: [PATCH 2/3] Install MCP client for LangGraph integration tests --- tox.ini | 1 + 1 file changed, 1 insertion(+) diff --git a/tox.ini b/tox.ini index 4d2d016a..552fe594 100644 --- a/tox.ini +++ b/tox.ini @@ -31,6 +31,7 @@ deps = pytest parameterized coverage + langgraph: mcp>=1.12,<2 mypy: mypy commands = coverage run --source=. --omit="**/tmp/**,./tmp/**/*" -m pytest -v {posargs} From 14cd8590849cbca5474dffcbf0cf1ed4523d303d Mon Sep 17 00:00:00 2001 From: Parallel Date: Tue, 20 Oct 2026 13:56:43 +0000 Subject: [PATCH 3/3] Require an explicit fetch objective across native tool schemas --- agentstack/_tools/parallel/__init__.py | 4 +-- docs/tools/tool/parallel.mdx | 3 ++- tests/tools/test_tool_parallel.py | 18 +++++++++---- tests/tools/test_tool_parallel_crewai.py | 34 ++++++++++++++++++++++++ 4 files changed, 51 insertions(+), 8 deletions(-) create mode 100644 tests/tools/test_tool_parallel_crewai.py diff --git a/agentstack/_tools/parallel/__init__.py b/agentstack/_tools/parallel/__init__.py index 8253941d..303f4b9c 100644 --- a/agentstack/_tools/parallel/__init__.py +++ b/agentstack/_tools/parallel/__init__.py @@ -70,12 +70,12 @@ def web_search(objective: str, search_queries: list[str]) -> str: return _run_tool('web_search', {'objective': objective, 'search_queries': search_queries}) -def web_fetch(urls: list[str], objective: str = '') -> str: +def web_fetch(urls: list[str], objective: str) -> str: """Read excerpts from up to 20 HTTP(S) pages, preserving source URLs. Args: urls: The web page URLs to read, such as sources returned by web_search. - objective: Optional description of relevant information, up to 200 characters. + objective: Description of relevant information, up to 200 characters; use an empty string for general excerpts. Returns: Page excerpts and their source URLs for the agent to analyze and cite. diff --git a/docs/tools/tool/parallel.mdx b/docs/tools/tool/parallel.mdx index e17ac60d..a63db720 100644 --- a/docs/tools/tool/parallel.mdx +++ b/docs/tools/tool/parallel.mdx @@ -18,7 +18,8 @@ agentstack tools add parallel This installs the MCP client and adds `web_search` and `web_fetch` to your agents. Use `--agent researcher` to restrict the tool to one agent. The tools return source URLs and excerpts for the agent to analyze and cite. `web_fetch` accepts up to 20 -HTTP(S) URLs and an optional objective of up to 200 characters. Requests time out +HTTP(S) URLs and an objective of up to 200 characters. Pass an empty objective +string for general excerpts. Requests time out after 30 seconds; service errors propagate to the framework without switching providers or retrying automatically. diff --git a/tests/tools/test_tool_parallel.py b/tests/tools/test_tool_parallel.py index 14994520..fb94163a 100644 --- a/tests/tools/test_tool_parallel.py +++ b/tests/tools/test_tool_parallel.py @@ -126,7 +126,9 @@ def test_search_fetch_and_followup_reach_consumer(project, server): app = dispatcher() with search_session(): assert 'https://example.com/research' in invoke(app) - assert 'Page excerpt' in invoke(app, 'web_fetch', {'urls': ['https://example.com/research']}) + assert 'Page excerpt' in invoke( + app, 'web_fetch', {'urls': ['https://example.com/research'], 'objective': ''} + ) invoke(app) assert [call['name'] for call in calls(server)] == ['web_search', 'web_fetch', 'web_search'] assert len({call['arguments']['session_id'] for call in calls(server)}) == 1 @@ -140,7 +142,7 @@ def test_new_and_concurrent_conversations_use_distinct_ids(project, server): def conversation(): with search_session(): invoke(app) - invoke(app, 'web_fetch', {'urls': ['https://example.com/research']}) + invoke(app, 'web_fetch', {'urls': ['https://example.com/research'], 'objective': ''}) with ThreadPoolExecutor(max_workers=2) as executor: list(executor.map(lambda _: conversation(), range(2))) @@ -161,7 +163,7 @@ def test_cli_starts_new_session_and_reuses_it_within_run(project, server): 'def run():\n' ' app = dispatcher()\n' ' invoke(app)\n' - ' invoke(app, "web_fetch", {"urls": ["https://example.com/research"]})\n' + ' invoke(app, "web_fetch", {"urls": ["https://example.com/research"], "objective": ""})\n' ) run_project() run_project() @@ -241,8 +243,8 @@ def test_structured_content_is_consumed(project, server): [ ('web_search', {'objective': '', 'search_queries': ['research']}), ('web_search', {'objective': 'research', 'search_queries': []}), - ('web_fetch', {'urls': ['file:///etc/passwd']}), - ('web_fetch', {'urls': []}), + ('web_fetch', {'urls': ['file:///etc/passwd'], 'objective': ''}), + ('web_fetch', {'urls': [], 'objective': ''}), ('web_fetch', {'urls': ['https://example.com'], 'objective': 'x' * 201}), ], ) @@ -252,6 +254,12 @@ def test_invalid_input_never_contacts_server(project, server, name, arguments): assert server[0] == [] +def test_missing_fetch_objective_fails_native_schema(project, server): + with pytest.raises(Exception, match='objective'): + invoke(dispatcher(), 'web_fetch', {'urls': ['https://example.com']}) + assert server[0] == [] + + def test_removed_tool_is_not_available_to_dispatcher(project, server): graph = StateGraph(MessagesState) graph.add_node('tools', ToolNode([])) diff --git a/tests/tools/test_tool_parallel_crewai.py b/tests/tools/test_tool_parallel_crewai.py new file mode 100644 index 00000000..37833c97 --- /dev/null +++ b/tests/tools/test_tool_parallel_crewai.py @@ -0,0 +1,34 @@ +"""Keep the native CrewAI tool schema consistent with the advertised inputs.""" + +import json + +import pytest + +import agentstack +from agentstack import conf +from agentstack._tools import parallel + + +pytest.importorskip('crewai') + + +def test_fetch_objective_is_explicit_and_accepts_empty_string(tmp_path, monkeypatch): + original_path = conf.PATH + (tmp_path / 'agentstack.json').write_text(json.dumps({'framework': 'crewai', 'tools': ['parallel']})) + conf.set_path(tmp_path) + seen = [] + + def result(name, arguments): + seen.append((name, arguments)) + return 'Source https://example.com: excerpt' + + monkeypatch.setattr(parallel, '_run_tool', result) + try: + _, fetch = agentstack.tools['parallel'] + with pytest.raises(ValueError, match='objective'): + fetch.args_schema.model_validate({'urls': ['https://example.com']}) + assert seen == [] + assert 'https://example.com' in fetch.run(urls=['https://example.com'], objective='') + assert seen == [('web_fetch', {'urls': ['https://example.com'], 'objective': None})] + finally: + conf.set_path(original_path)