# Agentic Mcp

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

- **Type:** MCP server
- **Install:** `agentstack add mcp-shy2593666979-agentic-mcp`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [Shy2593666979](https://agentstack.voostack.com/s/shy2593666979)
- **Installs:** 0
- **Category:** [AI & ML](https://agentstack.voostack.com/c/ai-and-ml)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [Shy2593666979](https://github.com/Shy2593666979)
- **Source:** https://github.com/Shy2593666979/agentic-mcp

## Install

```sh
agentstack add mcp-shy2593666979-agentic-mcp
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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工具管理方案
- **集成门槛高**: 现有解决方案配置复杂，学习成本高

## 🛠 安装

```bash
pip install agentic-mcp
```

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

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

### 依赖要求

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

## 📚 快速开始

### 1. MCPManager 基础使用

```python
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 智能代理使用

```python
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格式：

```python
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工具，可以通过工具格式转换实现：

```python
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)
```python
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
```python
from agentic_mcp.schemas import MCPWebsocketConfig

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

#### 3. Stdio
```python
from agentic_mcp.schemas import MCPStdioConfig

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

#### 4. HTTP
```python
from agentic_mcp.schemas import MCPStreamableHttpConfig

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

### 模型配置

```python
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)`: 获取工具调用的消息列表

## 🌟 高级用法

### 多服务器配置

```python
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](https://pypi.org/project/agentic-mcp/)
- [GitHub](https://github.com/Shy2593666979/agentic-mcp)
- [文档](https://github.com/Shy2593666979/agentic-mcp#readme)
- [问题反馈](https://github.com/Shy2593666979/agentic-mcp/issues)

---

**让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](https://github.com/Shy2593666979)
- **Source:** [Shy2593666979/agentic-mcp](https://github.com/Shy2593666979/agentic-mcp)
- **License:** MIT

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

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** yes
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/mcp-shy2593666979-agentic-mcp
- Seller: https://agentstack.voostack.com/s/shy2593666979
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.