Agent MCP

🚀 MCP Server and Client

This project consists of an MCP server and a client, aiming to facilitate local Web searches and integrate with external platforms.

💡 Core Functionality

• The MCP server is developed by referring to Antropic's Model Context Protocol Documentation. It uses Google SerpAPI as the search API for local Web searches.
• The MCP client can call both the locally deployed MCP server and the public MCP server service provided by the external platform Smithery. Smithery Platform

🚀 Quick Start

MCP Server

Refer to Antropic's Model Context Protocol Documentation to write a local Web search MCP server, using Google SerpAPI as the search API.

MCP Client

The local MCP client is designed to call both the locally deployed MCP server and the public MCP server service provided by the Smithery platform. Smithery Platform Refer to Antropic MCP Client Quick Start, modify the calling interface (originally using the Anthropic API) to the OpenAI interface, adjust the post - processing flow, and add some JSON validations. The testing results using Qwen2.5 series models and DeepSeek - V3 are quite stable.

💻 Usage Examples

Basic Usage

# Update on 2025.4.22
# custom_agents folder planning + Researcher Agent imitates in - depth research function, using MCP server
async with MCPServerStdio(
    params={"command": "python", "args": [LOCAL_SERVER_PATH]}
) as server:
    # List tools
    tools = await server.list_tools()
    print(f"Tools connected to the server: {tools}")
    # Plan the agent
    planning_agent = PlanningAgent(
        name="Planning Agent",
        model=model,
        handoffs=[
            ResearchAgent(
                name="Research Agent",
                model=model,
                mcp_servers=[server],
            )
        ],
    )
    # Run
    result_stream = Runner.run_streamed(
        starting_agent=planning_agent,
        input="What are the main focuses of AI in 2025? Give me some key points of trending academic papers and an overview of new technologies in 2025",
        max_turns=15,
    )
    async for event in result_stream.stream_events():
        if event.type == "agent_updated_stream_event":
            print(f"Agent updated: {event.new_agent.name}")
        elif event.type == "run_item_stream_event":
            if isinstance(event.item, ToolCallItem):
                print(f"Calling tool: {event.item.raw_item.name}, parameters: {event.item.raw_item.arguments}")
                await asyncio.sleep(0.05)
            elif isinstance(event.item, ToolCallOutputItem):
                text = json.loads(event.item.output)["text"]
                print(f"Tool call output: {text[:100]} ...")
                await asyncio.sleep(0.05)
            elif isinstance(event.item, MessageOutputItem):
                text = ItemHelpers.text_message_output(event.item)
                if text:
                    print(f"Running step: {text}")
                    await asyncio.sleep(0.05)
        elif event.type == "raw_response_event":
            try:
                if event.data.delta:
                    continue
            except Exception as e:
                print(f"Event data: {event.data}")