[**的Python实现服务器,使用了fastmcp和fastapi。
此仓库基于官方的MCP Python SDK仓库,目标是使用FastMCP创建一个Python的MCP服务器。项目包含了以下基本功能:
项目专注于实现一个简单的通过FastAPI和httpstream提供的MCP服务器。这种方法代表了创建MCP服务器的推荐方法。要探索其他实现形式和服务,请参考官方文档。
注意:对于生产部署,可流式传输的HTTP传输正在取代SSE传输。
from mcp.server.fastmcp import FastMCP
# 无状态服务器(无会话持久化)
mcp = FastMCP("StatelessServer", stateless_http=True)
您可以在FastAPI应用程序中挂载多个FastMCP服务器
# echo.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="EchoServer", stateless_http=True)
@mcp.tool(description="一个简单的回声工具")
def echo(message: str) -> str:
return f"回声: {message}"
# math.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="MathServer", stateless_http=True)
@mcp.tool(description="一个简单的加法工具")
def add_two(n: int) -> int:
return n + 2
# fast_api.py
import contextlib
from fastapi import FastAPI
from mcp.echo import echo
from mcp.math import math
# 创建一个组合生命周期来管理两个会话管理器
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
async with contextlib.AsyncExitStack() as stack:
await stack.enter_async_context(echo.mcp.session_manager.run())
await stack.enter_async_context(math.mcp.session_manager.run())
yield
app = FastAPI(lifespan=lifespan)
app.mount("/echo", echo.mcp.streamable_http_app())
app.mount("/math", math.mcp.streamable_http_app())
对于授权系统,使用了一个提供简单客户端凭证授权方法的包,名为mcp-oauth。这个包允许在运行MCP服务器的同时运行一个OAuth服务器。源代码可以在oauth_server.py找到。
# oauth_server.py
from mcp_oauth import (
OAuthServer,
SimpleAuthSettings,
AuthServerSettings,
)
from dotenv import load_dotenv
load_dotenv()
OAUTH_HOST = "127.0.0.1"
OAUTH_PORT = 9000
OAUTH_SERVER_URL = f"http://{OAUTH_HOST}:{OAUTH_PORT}"
def run_oauth_server():
server_settings: AuthServerSettings = AuthServerSettings(
host=OAUTH_HOST,
port=OAUTH_PORT,
server_url=f"{OAUTH_SERVER_URL}",
auth_callback_path=f"{OAUTH_SERVER_URL}/login",
)
auth_settings: SimpleAuthSettings = SimpleAuthSettings(
superusername=os.getenv("SUPERUSERNAME"),
superuserpassword=os.getenv("SUPERUSERPASSWORD"),
mcp_scope="user",
)
oauth_server: OAuthServer = OAuthServer(
server_settings=server_settings, auth_settings=auth_settings
)
oauth_server.run_starlette_server()
if __name__ == "__main__":
run_oauth_server()
要启动此服务器,可以在项目的根目录打开一个终端并执行:
python3 src/services/fast_mcp/private_server/oauth_server.py
一旦OAuth服务器运行起来,必须将其与MCP服务器集成,向MCP服务器提供OAuth服务器运行的地址:
def create_private_server(settings: ServerSettings = ServerSettings()) -> FastMCP:
token_verifier = IntrospectionTokenVerifier(
introspection_endpoint=settings.auth_server_introspection_endpoint,
server_url=str(settings.server_url),
validate_resource=settings.oauth_strict, # 只有当设置了--oauth-strict时才验证资源
)
mcp: FastMCP = FastMCP(
name="private-example-server",
instructions="此服务器专门用于用户资料数据的私有操作",
debug=True,
# RS模式的认证配置
token_verifier=token_verifier,
auth=AuthSettings(
issuer_url=settings.auth_server_url,
required_scopes=[settings.mcp_scope],
resource_server_url=settings.server_url,
),
)
MCP需要一个TokenVerifier,这里使用了由mcp_oauth包提供的一个简单的验证器。在这种情况下,settings.auth_server_url必须是OAuth服务器运行的地址,例如"http://127.0.0.1:9000"。有关进一步配置详情,请参阅代码在private_server/server.py。
此OAuth服务器使用基于凭证的身份验证系统(初始授权令牌获取)。您必须填写.env文件中的以下变量:
SUPERUSERNAME=user
SUPERUSERPASSWORD=password
要设置开发环境,请执行以下命令:
1. 安装项目依赖
pip install -r requirements.txt
2.1 开发模式启动服务器
uvicorn src.app:app --host 127.0.0.1 --port 8000 --reload
2.2 启动OAuth服务器
python3 src/services/fast_mcp/private_server/oauth_server.py
3. 验证服务器正确启动
要确认服务器正常运行,请打开浏览器并导航到地址http://127.0.0.1:8000。这应该重定向到一个用户帮助页面,该页面提供了如何使用服务器的指导。
4. 运行测试
python tests/run.py
该项目可以使用Docker Compose运行:
docker compose -f docker-compose.yml up -d --build
要验证此服务器的正确操作,建议安装mcp-llm-client包,并根据以下步骤创建一个基于它的项目:
⚠️ 配置提示: 要使用此聊天与LLM一起使用,需要OpenAI API密钥。如果您没有,可以通过遵循官方OpenAI页面上的说明来创建它。
1. 服务器部署
按照部署部分提供的说明部署此服务器。这一步至关重要,因为服务器必须在本地或云服务器上运行。一旦服务器部署完成,就可以通过MCP客户端使用它。
<!-- **2. 安装包** ```shell pip install mcp-llm-client ``` -->2. 从GitHub克隆模板
从GitHub克隆一个模板,该模板提供了一个简单的基础来使用MCP客户端:
# 克隆仓库
git clone https://github.com/rb58853/template_mcp_llm_client.git
# 切换到项目目录
cd template_mcp_llm_client
# 安装依赖
pip install -r requirements.txt
3. 添加服务器到配置
在克隆的项目中,在根目录下找到config.json文件,并在mcp_servers对象内添加以下配置:
{
"mcp_servers": {
"example_public_server": {
"transport": "httpstream",
"httpstream-url": "http://127.0.0.1:8000/public-example-server/mcp",
"name": "example-public-server",
"description": "示例公共服务器。"
},
"example_private_mcp": {
"transport": "httpstream",
"httpstream-url": "http://127.0.0.1:8000/private-example-server/mcp",
"name": "example-private-server",
"description": "需要OAuth的示例私有服务器。",
"auth": {
"required": true,
"post_body": {
"username": "user",
"password": "password"
}
}
}
}
}
💡 提示: 一旦服务器部署完成,您可以访问其根URL以获取帮助。这一节提供了将服务器添加到MCP客户端的确切配置。例如,打开
http://127.0.0.1:8000在浏览器中将重定向到帮助页面。
4. 执行
按照克隆项目中的readme.md文件中的说明,使用此MCP服务器运行本地聊天。通常,这可以通过在控制台中运行以下命令来完成:
# 运行应用(在设置OPENAI-API-KEY并添加服务器到配置后)
python3 main.py
要了解如何更详细地使用此MCP客户端,请参阅其官方仓库。
MIT许可证。详见license。