Agentic Mcp

Agentic MCP

> 为AI Agent提供MCP协议工具调度和管理的Python SDK ⭐

💡 为什么选择 Agentic MCP？

LangChain 生态系统无疑是构建AI Agent的优秀选择：新手只需几十行代码即可构建简单Agent，拥有丰富的生态包括 大模型厂商（OpenAI、Anthropic）、向量数据库（Milvus、Chroma）、沙箱（SandBox）等工具，生态完善且功能强大。

但随着AI框架的快速发展，一些问题逐渐显现：

• 🔄 版本依赖冲突: LangChain各生态包版本迭代不一致，容易产生依赖冲突
• 🎯 定制化过度复杂: 对于特定需求，使用完整框架显得过于臃肿
• ⚡ 轻量化需求: 希望保留LangChain优秀设计理念，但需要更轻量的解决方案

Agentic MCP 应运而生 - 专注于MCP协议的轻量级工具调度器，既汲取了LangChain的设计精髓，又避免了生态复杂性问题。

📖 项目简介

Agentic MCP 是一个专门为AI Agent设计的MCP（Model Context Protocol）工具调度器。它提供了一个不依赖LangChain的轻量级解决方案，专注于MCP工具的管理和调用，为构建MCP Agent提供了优化的实现方式。

🚀 核心特性

• 🔧 统一工具管理: 支持多种MCP传输协议（SSE、WebSocket、Stdio、HTTP）
• 🤖 智能代理集成: 与OpenAI等LLM无缝对接，支持并行工具调用
• ⚡ 异步流式处理: 原生支持异步操作和流式响应
• 🧩 轻量独立: 无需依赖LangChain，专注MCP工具调度
• 📦 开箱即用: 简单配置即可开始使用

🎯 解决的问题

• 依赖复杂: 避免LangChain生态的版本依赖问题，提供独立解决方案
• 工具管理复杂: 缺乏统一的多服务器MCP工具管理方案
• 集成门槛高: 现有解决方案配置复杂，学习成本高

🛠 安装

pip install agentic-mcp

如果 PyPI 安装失败，可以直接从 GitHub 安装：

pip install git+https://github.com/Shy2593666979/agentic-mcp.git

依赖要求

• Python >= 3.10
• mcp >= 1.10.0
• openai >= 1.12.0

📚 快速开始

1. MCPManager 基础使用

import asyncio
from agentic_mcp import MCPManager
from agentic_mcp.schemas import MCPSSEConfig

async def main():
    # 配置MCP服务
    gaode_config = MCPSSEConfig(
        server_name="高德地图",
        url="https://mcp.api-inference.modelscope.net/77df8a09751e4c/sse"
    )

    # 创建MCP管理器
    manager = MCPManager(mcp_configs=[gaode_config])

    # 获取可用工具列表
    tools = await manager.get_mcp_tools()
    print(f"可用工具: {[tool.name for tool in tools]}")

    # 查看工具详情 (用来展示到前端的信息)
    tools_info = await manager.show_mcp_tools()
    print(f"工具详情: {tools_info}")

    # 调用工具
    result = await manager.call_mcp_tools([
        {
            "tool_name": "maps_weather",
            "tool_args": {"city": "北京"}
        }
    ])
    print(f"调用结果: {result}")

if __name__ == "__main__":
    asyncio.run(main())

2. MCPAgent 智能代理使用

import asyncio
from agentic_mcp import MCPAgent
from agentic_mcp.schemas import ModelConfig, MCPSSEConfig

# 全局日志配置 (开启可查看工具调用的日志)
# import logging
# logging.basicConfig(
#     level=logging.INFO,  # 日志级别：DEBUG/INFO/WARNING/ERROR
#     format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"  # 日志格式
# )

async def main():
    # 配置语言模型
    model_config = ModelConfig(
        api_key="your-api-key",
        base_url="https://api.openai.com/v1",  # 或其他兼容接口
        model="gpt-4",
        model_kwargs={
            "temperature": 0.7,
            "parallel_tool_calls": True  # 一些模型必须通过该参数才能并行调用工具，点名qwen3-8b
        }
    )
    
    # 配置MCP服务
    mcp_config = MCPSSEConfig(
        server_name="高德地图",
        url="https://mcp.api-inference.modelscope.net/77df8a09751e4c/sse"
    )
    
    # 创建智能代理
    agent = MCPAgent(
        model_config=model_config,
        mcp_configs=[mcp_config]
    )
    
    # 流式对话
    async for chunk in agent.astream("帮我查一下北京到上海的路线"):
        print(chunk.content, end="")
    
    # 或者一次性获取结果
    response = await agent.ainvoke("北京今天天气怎么样？")
    print(response.content)
    
    # 又或者只想要获取到工具调用的Message
    call_messages = await agent.make_function_call_messages("北京今天的天气如何？")
    print(call_messages)
    
    
if __name__ == "__main__":
    asyncio.run(main())

3. 与 LangChain 生态集成

如果您想在现有的LangChain项目中使用MCP工具，可以通过MCPManager获取工具后转换为LangChain格式：

import asyncio
from typing import cast
from agentic_mcp import MCPManager
from agentic_mcp.schemas import MCPSSEConfig

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import StructuredTool
from langchain_openai import ChatOpenAI

async def main():
    # 配置LangChain LLM
    llm = ChatOpenAI(
        model="gpt-4",
        api_key="your-api-key",
        base_url="https://api.openai.com/v1"
    )

    # 配置MCP服务
    gaode_config = MCPSSEConfig(
        server_name="高德地图",
        url="https://mcp.api-inference.modelscope.net/77df8a09751e4c/sse",
    )

    # 创建MCP管理器
    mcp_manager = MCPManager(mcp_configs=[gaode_config])

    # 获得MCP服务的工具
    mcp_tools = await mcp_manager.get_mcp_tools()

    # 转成LangChain生态的格式
    llm_with_tools = llm.bind_tools([StructuredTool(**tool.model_dump()) for tool in mcp_tools])
    messages = [HumanMessage(content="北京的天气如何啊？")]

    # LLM推理并获取工具调用
    response = llm_with_tools.invoke(messages)
    response = cast(AIMessage, response)
    messages.append(response)

    # 执行工具调用
    if response.tool_calls:
        for tool_call in response.tool_calls:
            tool_name = tool_call.get("name")
            tool_args = tool_call.get("args")
            tool_id = tool_call.get("id")
            
            # 找到对应的MCP工具并执行
            for tool in mcp_tools:
                if tool.name == tool_name:
                    tool_result = await tool.coroutine(**tool_args)
                    messages.append(ToolMessage(
                        content=str(tool_result), 
                        name=tool_name, 
                        tool_call_id=tool_id
                    ))
                    break

    # 获取最终响应
    async for chunk in llm.astream(messages):
        print(chunk.content, end="")

if __name__ == "__main__":
    asyncio.run(main())

4. 与原生OpenAI客户端集成

如果您希望保持原生OpenAI API的调用方式，同时使用MCP工具，可以通过工具格式转换实现：

import asyncio
import json
from openai import OpenAI

from agentic_mcp.mcp.manager import MCPManager
from agentic_mcp.schemas.mcp import MCPSSEConfig
from agentic_mcp.utils.function import mcp_tool_to_args_schema

async def main():
    # 配置MCP服务
    gaode_config = MCPSSEConfig(
        server_name="高德地图",
        url="https://mcp.api-inference.modelscope.net/77df8a09751e4c/sse",
    )

    # 创建MCP管理器
    mcp_manager = MCPManager(mcp_configs=[gaode_config])

    # 获取MCP工具并转换为OpenAI格式
    mcp_tools = await mcp_manager.get_mcp_tools()
    openai_tools = mcp_tool_to_args_schema(mcp_tools)

    # 初始化OpenAI客户端
    client = OpenAI(
        api_key="your-api-key",
        base_url="https://api.openai.com/v1",
    )

    # 发起对话请求
    messages = [{"role": "user", "content": "北京天气如何啊？"}]
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=openai_tools
    )

    # 找到该工具对应的具体工具
    def find_mcp_tool(tool_name):
        for tool in mcp_tools:
            if tool.name == tool_name:
                return tool
        return None

    # 将AI回复添加到消息历史
    messages.append(response.choices[0].message)
    
    # 如果AI决定调用工具，执行工具调用
    if response.choices[0].message.tool_calls:
        for tool_call in response.choices[0].message.tool_calls:
            tool_name = tool_call.function.name
            tool_args = tool_call.function.arguments
            tool_id = tool_call.id

            # 调用对应的MCP工具
            mcp_tool = find_mcp_tool(tool_name)
            tool_result = await mcp_tool.coroutine(**json.loads(tool_args))

            # 将工具执行结果添加到消息历史
            messages.append({
                "name": tool_name,
                "role": "tool", 
                "content": str(tool_result),
                "tool_call_id": tool_id
            })

    # 基于工具调用结果生成最终回复
    for chunk in client.chat.completions.create(
        model="gpt-4", 
        messages=messages, 
        stream=True
    ):
        print(chunk.choices[0].delta.content, end="")

if __name__ == "__main__":
    asyncio.run(main())

🔧 配置说明

支持的传输协议

1. SSE (Server-Sent Events)

from agentic_mcp.schemas import MCPSSEConfig

config = MCPSSEConfig(
    server_name="服务名称",
    url="https://your-mcp-server.com/sse",
    headers={"Authorization": "Bearer token"},  # 可选
    timeout=30.0  # 可选
)

2. WebSocket

from agentic_mcp.schemas import MCPWebsocketConfig

config = MCPWebsocketConfig(
    server_name="服务名称",
    url="wss://your-mcp-server.com/ws"
)

3. Stdio

from agentic_mcp.schemas import MCPStdioConfig

config = MCPStdioConfig(
    server_name="服务名称",
    command="python",
    args=["/path/to/your/mcp_server.py"]
)

4. HTTP

from agentic_mcp.schemas import MCPStreamableHttpConfig

config = MCPStreamableHttpConfig(
    server_name="服务名称",
    url="https://your-mcp-server.com/mcp"
)

模型配置

from agentic_mcp.schemas import ModelConfig

config = ModelConfig(
    api_key="your-api-key",
    base_url="https://api.openai.com/v1",
    model="gpt-4",
    model_kwargs={
        "temperature": 0.7,
        "max_tokens": 1000,
        "parallel_tool_calls": True
    }
)

📋 API 文档

MCPManager

主要方法：

• get_mcp_tools(): 获取所有可用工具
• show_mcp_tools(): 查看工具详细信息
• call_mcp_tools(tools_info): 调用指定工具

MCPAgent

主要方法：

• ainvoke(message): 异步调用，返回完整响应
• astream(message): 异步流式调用，返回响应块迭代器
• make_function_call_messages(message): 获取工具调用的消息列表

🌟 高级用法

多服务器配置

from agentic_mcp import MCPAgent
from agentic_mcp.schemas.mcp import MCPSSEConfig, MCPWebsocketConfig
from agentic_mcp.schemas.llm import ModelConfig

configs = [
    MCPSSEConfig(
        server_name="地图服务",
        url="https://map-service.com/sse"
    ),
    MCPWebsocketConfig(
        server_name="天气服务", 
        url="wss://weather-service.com/ws"
    )
]

# 配置语言模型
model_config = ModelConfig(
    api_key="your-api-key",
    base_url="https://api.openai.com/v1",  # 或其他兼容接口
    model="gpt-4",
    model_kwargs={
        "temperature": 0.7,
        "parallel_tool_calls": True  # 一些模型必须通过该参数才能并行调用工具，点名qwen3-8b
    }
)

agent = MCPAgent(
    model_config=model_config,
    mcp_configs=configs
)

🤝 贡献指南

欢迎提交Issue和Pull Request来帮助改进项目！

1. Fork 本项目
2. 创建功能分支 (git checkout -b feature/AmazingFeature)
3. 提交更改 (git commit -m 'Add some AmazingFeature')
4. 推送到分支 (git push origin feature/AmazingFeature)
5. 开启 Pull Request

📄 许可证

本项目采用 MIT 许可证 - 查看 [LICENSE](LICENSE) 文件了解详情。

🔗 相关链接

• PyPI
• GitHub
• 文档
• 问题反馈

---

让AI Agent轻松连接MCP生态 🚀

Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: Shy2593666979
• Source: Shy2593666979/agentic-mcp
• License: MIT

Install and usage instructions live in the source repository linked above.