跳转到主要内容
可立即部署的实际 MCP 服务器示例。每个示例都展示了凭证配置、工具实现以及服务器配置。
## GitHub + Supabase
一个具有多个 Connection 的 MCP 服务器，公开 GitHub 和 Supabase 工具。
### 环境
# .env
DEDALUS_API_KEY=dsk-live-...

# Supabase
SUPABASE_URL=https://xxx.supabase.co
SUPABASE_SECRET_KEY=eyJ...

# GitHub
GITHUB_TOKEN=ghp_...
GITHUB_BASE_URL=https://api.github.com

​
服务器

# server.py
import os
from dedalus_mcp import MCPServer
from dedalus_mcp.server import TransportSecuritySettings
from dedalus_mcp.auth import Connection, SecretKeys

# 定义连接（凭据由 Client 在运行时提供）
github = Connection(
    name="github",
    secrets=SecretKeys(token="GITHUB_TOKEN"),
    auth_header_format="token {api_key}"
)

supabase = Connection(
    name="supabase",
    secrets=SecretKeys(token="SUPABASE_SECRET_KEY"),
    auth_header_format="Bearer {api_key}"
)

def create_server() -> MCPServer:
    return MCPServer(
        name="example-dedalus-mcp",
        connections=[github, supabase],
        http_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
        streamable_http_stateless=True,
        authorization_server=os.getenv("DEDALUS_AS_URL", "https://as.dedaluslabs.ai"),
    )

​
GitHub 工具

# gh.py
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.types import ToolAnnotations

@dataclass(frozen=True)
class GhResult:
    """GitHub API 结果。"""
    success: bool
    data: Any = None
    error: str | None = None

@dataclass(frozen=True)
class GhUser:
    """GitHub 用户资料。"""
    login: str
    name: str | None
    bio: str | None
    public_repos: int
    followers: int

@dataclass(frozen=True)
class GhRepo:
    """GitHub 仓库摘要。"""
    name: str
    full_name: str
    stars: int
    language: str | None
    updated_at: str

async def _gh_request(method: HttpMethod, path: str, body: Any = None) -> GhResult:
    """执行 GitHub API 请求。"""
    ctx = get_context()
    resp = await ctx.dispatch("github", HttpRequest(method=method, path=path, body=body))
    if resp.success:
        return GhResult(success=True, data=resp.response.body)
    return GhResult(success=False, error=resp.error.message if resp.error else "请求失败")

@tool(
    description="获取已认证 GitHub 用户的资料",
    tags=["user", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_whoami() -> GhResult:
    result = await _gh_request(HttpMethod.GET, "/user")
    if result.success and result.data:
        user = GhUser(
            login=result.data["login"],
            name=result.data.get("name"),
            bio=result.data.get("bio"),
            public_repos=result.data.get("public_repos", 0),
            followers=result.data.get("followers", 0),
        )
        return GhResult(success=True, data=user)
    return result

@tool(
    description="列出已认证用户的仓库",
    tags=["repos", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_list_repos(per_page: int = 10) -> GhResult:
    result = await _gh_request(HttpMethod.GET, f"/user/repos?per_page={per_page}&sort=updated")
    if result.success and isinstance(result.data, list):
        repos = [
            GhRepo(
                name=r["name"],
                full_name=r["full_name"],
                stars=r.get("stargazers_count", 0),
                language=r.get("language"),
                updated_at=r.get("updated_at", ""),
            )
            for r in result.data
        ]
        return GhResult(success=True, data=repos)
    return result

​
X（Twitter）API

使用 OAuth 2.0 仅应用（App-Only）身份验证的只读 X 应用程序编程接口 API 工具。

​
环境

# .env
X_BEARER_TOKEN=AAAA...  # 仅限应用的 bearer 令牌
X_API_KEY=...           # 可选:用于用户上下文端点
X_API_KEY_SECRET=...    # 可选:用于用户上下文端点
免费套餐额度：每月最多读取 100 条推文、写入 500 条推文。用于生产环境时，建议选择 Basic（$100/月）或 Pro 套餐。

​
Connection

# x.py
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
from dedalus_mcp.types import ToolAnnotations

x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    auth_header_format="Bearer {api_key}",
)

DEFAULT_TWEET_FIELDS = "id,text,author_id,created_at,public_metrics"
DEFAULT_USER_FIELDS = "id,name,username,description,public_metrics"

@dataclass(frozen=True)
class XResult:
    """X API 结果。"""
    success: bool
    data: Any = None
    meta: dict | None = None
    error: str | None = None

@dataclass(frozen=True)
class XUser:
    """X 用户资料。"""
    id: str
    name: str
    username: str
    description: str | None
    followers_count: int
    following_count: int

@dataclass(frozen=True)
class XTweet:
    """X 推文。"""
    id: str
    text: str
    author_id: str
    created_at: str
    retweet_count: int
    like_count: int

async def _x_request(path: str) -> XResult:
    """执行 X API 请求。"""
    ctx = get_context()
    resp = await ctx.dispatch("x", HttpRequest(method=HttpMethod.GET, path=path))
    if resp.success:
        body = resp.response.body or {}
        return XResult(success=True, data=body.get("data"), meta=body.get("meta"))
    return XResult(success=False, error=resp.error.message if resp.error else "请求失败")

​
工具

@tool(
    description="通过用户名获取 X 用户",
    tags=["user", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_by_username(username: str) -> XResult:
    result = await _x_request(f"/2/users/by/username/{username}?user.fields={DEFAULT_USER_FIELDS}")
    if result.success and result.data:
        metrics = result.data.get("public_metrics", {})
        user = XUser(
            id=result.data["id"],
            name=result.data["name"],
            username=result.data["username"],
            description=result.data.get("description"),
            followers_count=metrics.get("followers_count", 0),
            following_count=metrics.get("following_count", 0),
        )
        return XResult(success=True, data=user)
    return result

@tool(
    description="搜索近期推文（最近 7 天）",
    tags=["search", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_search_recent(query: str, max_results: int = 10) -> XResult:
    from urllib.parse import quote
    max_results = max(10, min(100, max_results))
    result = await _x_request(
        f"/2/tweets/search/recent?query={quote(query)}&tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
    )
    if result.success and result.data:
        tweets = [
            XTweet(
                id=t["id"],
                text=t["text"],
                author_id=t["author_id"],
                created_at=t.get("created_at", ""),
                retweet_count=t.get("public_metrics", {}).get("retweet_count", 0),
                like_count=t.get("public_metrics", {}).get("like_count", 0),
            )
            for t in result.data
        ]
        return XResult(success=True, data=tweets, meta=result.meta)
    return result

@tool(
    description="获取用户的近期推文",
    tags=["tweet", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_tweets(user_id: str, max_results: int = 10) -> XResult:
    max_results = max(5, min(100, max_results))
    result = await _x_request(
        f"/2/users/{user_id}/tweets?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
    )
    if result.success and result.data:
        tweets = [
            XTweet(
                id=t["id"],
                text=t["text"],
                author_id=t["author_id"],
                created_at=t.get("created_at", ""),
                retweet_count=t.get("public_metrics", {}).get("retweet_count", 0),
                like_count=t.get("public_metrics", {}).get("like_count", 0),
            )
            for t in result.data
        ]
        return XResult(success=True, data=tweets, meta=result.meta)
    return result

​
Gmail（OAuth 2.0）

支持真正的 OAuth 2.0 用户身份验证的 Gmail MCP 服务器。
### 环境
# .env
OAUTH_ENABLED=true
OAUTH_AUTHORIZE_URL=https://accounts.google.com/o/oauth2/auth
OAUTH_TOKEN_URL=https://oauth2.googleapis.com/token
OAUTH_CLIENT_ID=your-client-id.apps.googleusercontent.com
OAUTH_CLIENT_SECRET=your-client-secret
OAUTH_SCOPES_AVAILABLE=https://www.googleapis.com/auth/gmail.readonly,https://www.googleapis.com/auth/gmail.modify
OAUTH_BASE_URL=https://gmail.googleapis.com

DEDALUS_API_KEY=dsk-live-...
DEDALUS_API_URL=https://api.dedaluslabs.ai
DEDALUS_AS_URL=https://as.dedaluslabs.ai

​
设置

  1. 启用 Gmail 应用程序编程接口 API：访问 Gmail API 并点击 “Enable”
  2. 创建 OAuth 凭据：前往 APIs & Services → Credentials，并执行以下操作：
    • 点击 “Create Credentials” → “OAuth client ID”
    • Application type 选择 “Web application”
    • 为你的部署添加授权重定向 URI
    • 将 Client ID 和 Client Secret 复制到你的 .env 中
  3. 配置授权同意界面：使用上面列出的 Gmail 权限范围（scopes）设置 OAuth 授权同意界面
### 服务器
# server.py
import os
from dedalus_mcp import MCPServer
from dedalus_mcp.server import TransportSecuritySettings
from gmail import gmail, gmail_tools

def create_server() -> MCPServer:
    return MCPServer(
        name="gmail-mcp",
        connections=[gmail],
        http_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
        streamable_http_stateless=True,
        authorization_server=os.getenv("DEDALUS_AS_URL", "https://as.dedaluslabs.ai"),
    )

async def main() -> None:
    server = create_server()
    server.collect(*gmail_tools)
    await server.serve(port=8080)

​
Connection

# gmail.py
import os
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, OAuthConfig
from dedalus_mcp.types import ToolAnnotations

gmail = Connection(
    name="gmail",
    oauth=OAuthConfig(
        client_id=os.getenv("OAUTH_CLIENT_ID"),
        client_secret=os.getenv("OAUTH_CLIENT_SECRET"),
        authorize_url=os.getenv("OAUTH_AUTHORIZE_URL"),
        token_url=os.getenv("OAUTH_TOKEN_URL"),
        scopes=os.getenv("OAUTH_SCOPES_AVAILABLE", "").split(","),
    ),
    auth_header_format="Bearer {access_token}",
)

@dataclass(frozen=True)
class GmailResult:
    """Gmail API 结果。"""
    success: bool
    data: Any = None
    error: str | None = None

@dataclass(frozen=True)
class GmailMessage:
    """Gmail 消息摘要。"""
    id: str
    thread_id: str
    snippet: str | None
    label_ids: list[str]

async def _gmail_request(method: HttpMethod, path: str) -> GmailResult:
    """执行 Gmail API 请求。"""
    ctx = get_context()
    resp = await ctx.dispatch("gmail", HttpRequest(method=method, path=path))
    if resp.success:
        return GmailResult(success=True, data=resp.response.body)
    return GmailResult(success=False, error=resp.error.message if resp.error else "Request failed")

​
工具

@tool(
    description="列出最近的邮件",
    tags=["email", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_list_messages(max_results: int = 10, query: str = "") -> GmailResult:
    path = f"/gmail/v1/users/me/messages?maxResults={max_results}"
    if query:
        path += f"&q={query}"
    result = await _gmail_request(HttpMethod.GET, path)
    if result.success and result.data:
        messages = [
            GmailMessage(
                id=m["id"],
                thread_id=m.get("threadId", ""),
                snippet=None,
                label_ids=[],
            )
            for m in result.data.get("messages", [])
        ]
        return GmailResult(success=True, data=messages)
    return result

@tool(
    description="根据 ID 获取邮件详情",
    tags=["email", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_get_message(message_id: str) -> GmailResult:
    result = await _gmail_request(HttpMethod.GET, f"/gmail/v1/users/me/messages/{message_id}")
    if result.success and result.data:
        msg = result.data
        message = GmailMessage(
            id=msg["id"],
            thread_id=msg.get("threadId", ""),
            snippet=msg.get("snippet"),
            label_ids=msg.get("labelIds", []),
        )
        return GmailResult(success=True, data=message)
    return result

​
运行示例

所有示例都遵循相同的流程： 
# 克隆仓库
git clone https://github.com/dedalus-labs/example-dedalus-mcp
cd example-dedalus-mcp

# 主示例（GitHub + Supabase）
cp .env.example .env
# 使用您的凭据编辑 .env 文件

uv sync
uv run python src/main.py

# 适用于 X
cd x-mcp
cp .env.example .env
uv sync
uv run python src/main.py

# 适用于 Gmail
cd gmail-mcp
cp .env.example .env
uv sync
uv run python src/main.py
服务器在 http://127.0.0.1:8080/mcp 上运行。
Last modified on April 11, 2026