Skip to content
This repository was archived by the owner on Jul 19, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down
85 changes: 85 additions & 0 deletions agent_harness.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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):
Comment thread
cursor[bot] marked this conversation as resolved.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

API receives uncapped num_results but display silently truncates

Low Severity

The num_results value is sent to the API as num_web_results without any cap (line 298), but results are locally truncated to max(1, min(num_results, 10)) (line 329). When a caller requests, say, 20 results, the API fetches and returns all 20 but only 10 are displayed. The cap applied to the display loop isn't applied to the API request parameter, causing wasted bandwidth and silently incomplete output that could mislead the LLM agent.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 98c655d. Configure here.

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 ""

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Empty snippets list prevents fallback to description field

Low Severity

When a result item has "snippets": [] (an empty list), isinstance(snippets, list) is True, so the code enters the list-join branch and produces an empty string. It never falls back to hit.get("description") in the else branch, even though a useful description may be available. The result is output with only a title and URL but no descriptive text, despite the API providing one. The empty-list case needs to fall through to the description fallback.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 98c655d. Configure here.

snippet = " ".join(snippet.split())
Comment thread
cursor[bot] marked this conversation as resolved.
line = f"{idx}. {title}"
if link:
line += f"\n {link}"
if snippet:
line += f"\n {snippet[:220]}"
lines.append(line)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Response parsing outside try-except can raise unhandled exceptions

Medium Severity

The try/except block only covers the HTTP request and json.loads call (lines 309–313), but all subsequent response parsing (lines 315–343) is unprotected. If json.loads returns a non-dict value (e.g., None or a string from an unusual API response), payload.get("results") raises an AttributeError. Likewise, if individual items in the hits list are not dicts, hit.get("title") crashes the same way. The PR claims "no crash" error handling, but this gap allows unhandled exceptions to propagate to the caller.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 98c655d. Configure here.


return "\n".join(lines)

get_registry().register(name=name, description=description, callable=_you_search)
return name


@dataclass
class HarnessConfig:
"""Configuration for the harness"""
Expand Down
56 changes: 56 additions & 0 deletions tests/test_you_search_tool.py
Original file line number Diff line number Diff line change
@@ -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