AI导哈

MCP服务器实现

一个基于Python和FastAPI的Model Context Protocol (MCP)服务器实现，提供HTTP流传输和OAuth授权功能，适用于MCP客户端测试和开发。

  • by rb58853
  • Python
  • 本地部署
  • 8 星标
  • 更新于
在 GitHub 查看 rb58853/simple-mcp-server

使用说明

简单的Python MCP服务器

python实现的 模型上下文协议（MCP） 服务器与 fastmcp 和 fastapi.

目录

概述

此存储库基于官方的MCP Python SDK存储库，目的是使用FastMCP在Python中创建MCP服务器。该项目包括以下基本功能：

  • 从基础知识出发，以易于理解的方式促进对模型上下文协议（MCP）的理解和使用
  • 为MCP客户提供测试平台
  • 将服务器与FastAPI集成，并将其作为可流式传输的HTTP服务提供，保持服务和客户端之间的明确分离

该项目侧重于实现一个简单的MCP服务器，该服务器通过FastAPI和httpstream提供服务。这种方法代表了创建MCP服务器的推荐方法。要探索其他实现形式和服务器服务，建议咨询 官方文件.

运输

可流式HTTP传输

注意：在生产部署中，流式HTTP传输正在取代SSE传输。

from mcp.server.fastmcp import FastMCP
# Stateless server (no session persistence)
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="A simple echo tool")
def echo(message: str) -> str:
    return f"Echo: {message}"

# math.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(name="MathServer", stateless_http=True)

@mcp.tool(description="A simple add tool")
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


# Create a combined lifespan to manage both session managers
@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())

自动化（OAuth）

对于授权系统，使用了一个提供简单客户端凭据授权方法的包，称为 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

MCP集成

OAuth服务器运行后，必须通过向MCP服务器提供OAuth服务器运行的地址，将其与MCP服务器集成：

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,  # Only validate when --oauth-strict is set
    )

    mcp: FastMCP = FastMCP(
        name="private-example-server",
        instructions="This server specializes in private operations of user profiles data",
        debug=True,
        # Auth configuration for RS mode
        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.验证服务器是否正确启动

要确认服务器是否正常运行，请打开web浏览器并导航到该地址 http://127.0.0.1:8000。这应该重定向到用户帮助页面，该页面提供了如何使用服务器的指导。

4.运行测试

python tests/run.py

Docker部署

该项目可以使用Docker Compose运行：

docker compose -f docker-compose.yml up -d --build

用例

要验证此服务器的正确操作，建议安装 mcp-llm-client 按照以下步骤打包并创建基于它的项目：

⚠️ 配置说明： 要将此聊天与LLM一起使用，需要一个OpenAI API密钥。如果您没有，可以按照上的说明创建它 OpenAI官方页面.

1.服务器部署

根据中提供的说明部署此服务器 部署 部分。这一步至关重要，因为服务器必须在本地或云服务器上运行。一旦部署了服务器，就可以通过MCP客户端使用它。

2.从GitHub克隆模板

从GitHub克隆一个模板，为使用MCP客户端提供一个简单的基础：

# clone repo
git clone https://github.com/rb58853/template_mcp_llm_client.git

# change to project dir
cd template_mcp_llm_client

# install dependencies
pip install -r requirements.txt

3.将服务器添加到配置中

在克隆的项目中，找到 config.json 在根目录下的文件中添加以下配置 mcp服务器 对象：

{
    "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 public server."
        },
        "example_private_mcp": {
            "transport": "httpstream",
            "httpstream-url": "http://127.0.0.1:8000/private-example-server/mcp",
            "name": "example-private-server",
            "description": "Example private server with oauth required.",
            "auth": {
                "required": true,
                "post_body": {
                    "username": "user",
                    "password": "password"
                }
            }
        }
    }
}

💡 提示： 部署服务器后，您可以访问其根URL以获取帮助。本节提供了将服务器添加到MCP客户端所需的确切配置。例如，打开 http://127.0.0.1:8000 在浏览器中，将重定向到帮助页面。

4.执行

按照中的说明进行操作 readme.md 使用此MCP服务器运行本地聊天的克隆项目的文件。通常，这是通过在控制台中运行以下命令来完成的：

# Run app (after set OPENAI-API-KEY and add servers to config)
python3 main.py

参考文献

有关使用此MCP客户端的更多详细信息，请参阅 其官方存储库.

许可证

MIT许可证。看 license.

相关 MCP