diff --git a/README.md b/README.md index ea905e6..79a91ea 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,34 @@ Tools are automatically: - Wrapped for Claude using `@tool` + in-process MCP server - Available to both providers without duplication +#### Optional provider: You.com web search + +If your agents need live web results, register the built-in You.com adapter as a tool: + +```python +import os +from agent_harness import AgentHarness, HarnessConfig, register_you_com_search_tool + +os.environ["YOUCOM_API_KEY"] = "your_you_com_api_key" +register_you_com_search_tool(name="you_search") + +config = HarnessConfig( + system_prompt="Use web search when the user asks for fresh information.", + tool_names=["you_search"], +) + +# The tool call shape is: +# you_search(query: str, num_results: int = 5) +``` + +Environment variables: +- `YOUCOM_API_KEY` (required) + +Fallback/error behavior: +- Missing API key returns a non-fatal setup message (no crash) +- Network/API failures return a concise error string to the agent +- Empty result sets return `No web results found.` + ### Configuration `HarnessConfig` provides unified configuration: @@ -324,6 +352,14 @@ def register_tool( ) -> Callable def get_registry() -> ToolRegistry + +def register_you_com_search_tool( + name: str = "you_search", + description: str = "Search the web and return summarized results.", + api_key_env: str = "YOUCOM_API_KEY", + endpoint: str = "https://api.ydc-index.io/search", + timeout_sec: float = 20.0, +) -> str ``` ## Implementation Details diff --git a/agent_harness.py b/agent_harness.py index 7b3d72e..c4ff1ab 100644 --- a/agent_harness.py +++ b/agent_harness.py @@ -7,9 +7,13 @@ import asyncio import inspect +import json import logging import os import threading +import urllib.error +import urllib.parse +import urllib.request import uuid from abc import ABC, abstractmethod from dataclasses import dataclass, field @@ -263,6 +267,87 @@ def get_registry() -> ToolRegistry: return _global_registry +def register_you_com_search_tool( + name: str = "you_search", + description: str = "Search the web and return summarized results.", + api_key_env: str = "YOUCOM_API_KEY", + endpoint: str = "https://api.ydc-index.io/search", + timeout_sec: float = 20.0, +) -> str: + """ + Register a You.com web-search tool in the global registry. + + The registered tool accepts: + - query: str (required) + - num_results: int = 5 + + Returns the registered tool name. + """ + + def _you_search(query: str, num_results: int = 5) -> str: + api_key = os.getenv(api_key_env) + if not api_key: + return ( + f"You.com search is not configured. Set {api_key_env} and retry. " + "Returning no web results." + ) + + if not query.strip(): + return "You.com search query is empty." + + url = endpoint + "?" + urllib.parse.urlencode({"query": query, "num_web_results": num_results}) + request = urllib.request.Request( + url, + headers={ + "X-API-Key": api_key, + "Accept": "application/json", + "User-Agent": "agent-harness-you-search/1.0", + }, + method="GET", + ) + + try: + with urllib.request.urlopen(request, timeout=timeout_sec) as response: + payload = json.loads(response.read().decode("utf-8")) + except (urllib.error.HTTPError, urllib.error.URLError, json.JSONDecodeError, OSError) as exc: + return f"You.com search request failed ({type(exc).__name__}): {exc}" + + results = payload.get("results") + if isinstance(results, dict): + hits = results.get("web") or [] + elif isinstance(results, list): + hits = results + else: + hits = payload.get("hits") or [] + + if not isinstance(hits, list): + return "No web results found." + if not hits: + return "No web results found." + + lines = [] + for idx, hit in enumerate(hits[: max(1, min(num_results, 10))], start=1): + title = hit.get("title") or "Untitled" + link = hit.get("url") or hit.get("link") or "" + snippets = hit.get("snippets") + if isinstance(snippets, list): + snippet = " ".join(str(item) for item in snippets if item) + else: + snippet = hit.get("snippet") or hit.get("description") or "" + snippet = " ".join(snippet.split()) + line = f"{idx}. {title}" + if link: + line += f"\n {link}" + if snippet: + line += f"\n {snippet[:220]}" + lines.append(line) + + return "\n".join(lines) + + get_registry().register(name=name, description=description, callable=_you_search) + return name + + @dataclass class HarnessConfig: """Configuration for the harness""" diff --git a/tests/test_you_search_tool.py b/tests/test_you_search_tool.py new file mode 100644 index 0000000..a685c52 --- /dev/null +++ b/tests/test_you_search_tool.py @@ -0,0 +1,56 @@ +"""Tests for You.com search tool registration.""" + +import io +import json + +from agent_harness import get_registry, register_you_com_search_tool + + +def test_you_search_tool_requires_api_key(monkeypatch): + get_registry().clear() + monkeypatch.delenv("YOUCOM_API_KEY", raising=False) + + register_you_com_search_tool(name="you_search_test") + tool = get_registry().get("you_search_test") + + result = tool.callable("latest ai agent frameworks") + assert "not configured" in result.lower() + + +def test_you_search_tool_formats_results(monkeypatch): + get_registry().clear() + monkeypatch.setenv("YOUCOM_API_KEY", "test-key") + + class FakeResponse: + def __enter__(self): + payload = { + "results": { + "web": [ + { + "title": "Agent blog", + "url": "https://example.com/agent", + "description": "Generic page summary.", + "snippets": ["Useful update.", "Fresh context."], + } + ] + } + } + self._buf = io.BytesIO(json.dumps(payload).encode("utf-8")) + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return self._buf.read() + + monkeypatch.setattr("urllib.request.urlopen", lambda *args, **kwargs: FakeResponse()) + + register_you_com_search_tool(name="you_search_test") + tool = get_registry().get("you_search_test") + + result = tool.callable("agent tooling", num_results=1) + assert "1. Agent blog" in result + assert "https://example.com/agent" in result + assert "Useful update. Fresh context." in result + assert "Generic page summary." not in result