AI导哈

OAuth 2.0工具服务器

一个基于FastMCP构建的OAuth 2.0保护工具服务器，提供与Google API集成的功能。

  • by frnnk
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 frnnk/assistant-mcp

使用说明

助理MCP

MCP（模型上下文协议）服务器模板，为代理客户端提供OAuth 2.0保护的工具。使用FastMCP构建，目前与Google API集成。

组件

组件 位置 目的
FastMCP服务器 src/main.py 入口点、工具定义、OAuth路由
OAuth门 src/auth/oauth_gate.py 令牌验证和OAuth流启动
OAuthToolApp src/mcp_tools/auth_tool_app.py OAuth保护工具的基类
谷歌提供商 src/auth/providers/google_provider.py 谷歌OAuth 2.0实现
日历工具 src/mcp_tools/google/calendar.py Google日历API包装器
装饰师 src/utils/decorators.py 作用域、重试、OAuth错误处理

OAuth流程步骤

步骤 说明
1 客户端调用MCP工具（例如。， list_calendars)
2 @mcp_oauth_handler 装饰器包装工具功能
3 OAuthToolApp.run_method 传递给 ensure_auth
4 ensure_auth 通过提供程序检查有效令牌
5 如果没有代币，则加薪 OAuthRequiredError 随着 elicitation_id
6 装饰师转换为 UrlElicitationRequiredError 带有身份验证URL
7 客户端将用户重定向到 /auth/connect/{elicitation_id}
8 服务器重定向到Google OAuth同意屏幕
9 同意后，谷歌重定向到 /auth/callback/{elicitation_id}
10 服务器将代码交换为令牌并存储
11 客户端使用有效令牌重试原始工具调用

项目结构

assistant-mcp/
├── .env                       # Environment configuration
├── .gitignore
├── .python-version            # Python 3.13
├── pyproject.toml             # Dependencies and metadata
├── uv.lock                    # Dependency lock file
├── CLAUDE.md                  # Project documentation
├── README.md
│
└── src/
    ├── main.py                # FastMCP server entry point, tool definitions
    │
    ├── auth/
    │   ├── oauth_gate.py      # OAuth flow management, token elicitation
    │   ├── providers/
    │   │   ├── provider.py        # Abstract OAuthProvider interface
    │   │   ├── google_provider.py # Google OAuth implementation
    │   │   └── provider_registry.py # Provider lookup by name
    │   └── tokens/
    │       ├── auth_token.py      # Abstract token interface
    │       └── google_token.py    # Google token implementation
    │
    ├── mcp_tools/
    │   ├── auth_tool_app.py   # Base class for OAuth-protected tools
    │   └── google/
    │       └── calendar.py    # Google Calendar tool implementations
    │
    ├── utils/
    │   ├── decorators.py      # @tool_scope_factory, @tool_retry_factory, @mcp_oauth_handler
    │   └── errors.py          # Custom exceptions (OAuthRequiredError, etc.)
    │
    └── db/
        └── db.py              # Database utilities

安装

# Clone the repository
git clone <repo-url>
cd assistant-mcp

# Install dependencies with uv
uv sync

配置

创建一个 .env 文件包含：

GOOGLE_SECRETS_PATH=./secrets/client_secret.json
GOOGLE_LOCAL_TOKEN_PATH=./secrets/token.json
SERVER_HOST=127.0.0.1
SERVER_PORT=8000
变量 描述
GOOGLE_SECRETS_PATH Google OAuth客户端机密JSON文件的路径
GOOGLE_LOCAL_TOKEN_PATH OAuth令牌的存储路径
SERVER_HOST 服务器主机地址（默认值：127.0.0.1）
SERVER_PORT 服务器端口号（默认值：8000）

Google OAuth本地设置

对于本地设置，您需要创建一个Google Cloud项目并放置一个 secrets.json 文件 在该仓库的根目录下设置Google OAuth Provider。

跑步

uv run python src/main.py

服务器运行在 http://{SERVER_HOST}:{SERVER_PORT}/mcp 使用流式HTTP传输。

MCP工具

日历工具

工具 参数 输出 需要OAuth
list_calendars （无） calendar_id, name, description 是的
list_events calendar_id, start_time, duration_days event_id, name, start, end 是的
create_event calendar_id, start, name, duration_minutes, location, description event_id, event_details 是的
update_event calendar_id, event_id, start, name, duration_minutes, location, description event_id, event_details 是的

工具先决条件

工具 先决条件
list_calendars 没有
list_events calendar_id 从 list_calendars
create_event calendar_id 从 list_calendars
update_event calendar_id 从 list_calendars, event_id 从 list_events

默认值

  • calendar_id: 'primary' （如果未指定，则使用主日历）
  • duration_minutes: 30 （用于创建/更新）
  • duration_days: 7 （针对list_events）

OAuth流

当工具需要身份验证且不存在有效令牌时：

1. Client calls tool (e.g., list_calendars)
   ↓
2. OAuthToolApp.run_method invokes ensure_auth
   ↓
3. ensure_auth checks provider for valid token
   ↓
4. No token found → OAuthRequiredError with elicitation_id
   ↓
5. @mcp_oauth_handler converts to UrlElicitationRequiredError
   ↓
6. MCP client receives auth URL, displays to user
   ↓
7. User visits /auth/connect/{elicitation_id}
   ↓
8. Server redirects to Google OAuth consent
   ↓
9. User authorizes, Google redirects to /auth/callback/{elicitation_id}
   ↓
10. Server exchanges code for token, stores in JSON file
    ↓
11. User sees "You may close this tab"
    ↓
12. Client retries tool call → succeeds with valid token

API路线

GET/auth/connect/{启发_id}

通过重定向到Google OAuth同意屏幕来启动OAuth流。

请求：

curl -L http://127.0.0.1:8000/auth/connect/{elicitation_id}

答复： 302重定向到Google OAuth同意URL

GET/auth/recall/{启发_id}

OAuth回调处理程序。接收授权码并交换令牌。

请求： （来自谷歌重定向）

http://127.0.0.1:8000/auth/callback/{elicitation_id}?code=...&state=...

答复：

You may close this tab.

关键设计模式

工厂图案（装饰）

装饰器为工具方法创建专门的行为：

@tool_scope_factory(scopes=["https://www.googleapis.com/auth/calendar"])
@tool_retry_factory(error_message="Google Calendar error", retry_on=(HttpError,))
def list_calendars(self, *, token: GoogleToken, ctx: Dict[str, Any]):
    ...

策略模式（OAuthProvider接口）

抽象提供者接口允许多种OAuth实现：

class OAuthProvider(ABC):
    @abstractmethod
    def get_access_token(self, principal_id: str, scopes: Sequence[str]) -> Optional[OAuthToken]:
        ...

    @abstractmethod
    def generate_auth_url(self, scopes: Sequence[str], elicitation_id: str, ...) -> dict:
        ...

装饰师组成

工具函数组成多个装饰器：

@mcp.tool()
@mcp_oauth_handler("Authorization is required to access your Google Calendar.")
def list_calendars(ctx: Context):
    return calendar_tools.run_method('list_calendars', ctx=ctx)
  • @mcp.tool()：寄存器用作MCP工具
  • @mcp_oauth_handler：处理OAuth错误，转换为URL启发

关注点分离

  • main.py：仅工具定义和MCP布线
  • auth/：所有OAuth逻辑（提供者、令牌、流管理）
  • mcp_tools/：具有业务逻辑的工具实现
  • utils/：共享实用程序（装饰器、错误）

工具实施

要添加新的OAuth保护工具，请执行以下操作：

# In src/main.py
@mcp.tool()
@mcp_oauth_handler("Authorization message")
def my_tool(ctx: Context, param: str):
    """Docstring becomes tool description"""
    return tool_app.run_method('method_name', ctx=ctx, param=param)

# In tool app class
@tool_scope_factory(scopes=[...])
@tool_retry_factory(error_message=..., retry_on=(...))
def method_name(self, *, token: GoogleToken, ctx: Dict, param: str):
    # Implementation using token.present_creds()
    ...

相关

  • msg代理：使用LangGraph使用这些工具的配套MCP客户端项目
  • 测试客户端：提供了一个粗略的测试客户端来测试MCP服务器

相关 MCP