Function bocha_ai_search

tools/bocha_search_server.py:115–195

Search with Bocha AI Search, recognizes the semantics of search terms and additionally returns structured modal cards with content from vertical domains. Args: query: Search query (required) freshness: The time range for the search results. (Available options noLimit, oneYea

(
    query: str, freshness: str = "noLimit", count: int = 10
)

@server.tool()
async def bocha_ai_search(
    query: str, freshness: str = "noLimit", count: int = 10
) -> str:
    """Search with Bocha AI Search, recognizes the semantics of search terms
    and additionally returns structured modal cards with content from vertical domains.

    Args:
        query: Search query (required)
        freshness: The time range for the search results. (Available options noLimit, oneYear, oneMonth, oneWeek, oneDay. Default is noLimit)
        count: Number of results (1-50, default 10)
    """
    # Get API key from environment
    boch_api_key = os.environ.get("BOCHA_API_KEY", "")

    if not boch_api_key:
        return (
            "Error: Bocha API key is not configured. Please set the "
            "BOCHA_API_KEY environment variable."
        )

    # Endpoint
    endpoint = "https://api.bochaai.com/v1/ai-search?utm_source=bocha-mcp-local"

    try:
        payload = {
            "query": query,
            "freshness": freshness,
            "count": count,
            "answer": False,
            "stream": False,
        }

        headers = {
            "Authorization": f"Bearer {boch_api_key}",
            "Content-Type": "application/json",
        }

        async with httpx.AsyncClient() as client:
            response = await client.post(
                endpoint, headers=headers, json=payload, timeout=10.0
            )

            response.raise_for_status()
            response = response.json()
            results = []
            if "messages" in response:
                for message in response["messages"]:
                    content = {}
                    try:
                        content = json.loads(message["content"])
                    except (json.JSONDecodeError, TypeError):
                        content = {}

                    # 网页
                    if message["content_type"] == "webpage":
                        if "value" in content:
                            for item in content["value"]:
                                results.append(