Last updated
{
"name": "read_file",
"description": "读取指定路径的文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"}
},
"required": ["path"]
}
}{
"uri": "file://config.json",
"name": "配置文件",
"mimeType": "application/json"
}{
"name": "code_review",
"description": "代码审查提示模板",
"arguments": [
{"name": "code", "required": true}
]
}from mcp import Server
server = Server("my-service")
@server.tool()
async def my_tool(param: str) -> str:
"""工具描述"""
result = process(param)
return result
@server.resource("config://main")
async def get_config() -> str:
"""获取配置"""
return load_config()
if __name__ == "__main__":
server.run()from mcp.server.fastmcp import FastMCP
import httpx
# 初始化 MCP Server
mcp = FastMCP("github-server")
GITHUB_TOKEN = "your_token_here" # 实际使用时从环境变量读取
# ============ Tools：可执行的操作 ============
@mcp.tool()
async def search_repos(query: str, language: str = None) -> str:
"""搜索 GitHub 仓库
Args:
query: 搜索关键词
language: 可选，编程语言过滤
"""
search_query = f"{query} language:{language}" if language else query
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.github.com/search/repositories",
params={"q": search_query, "per_page": 5},
headers={"Authorization": f"token {GITHUB_TOKEN}"}
)
repos = resp.json().get("items", [])
return "\n".join([
f"- {r['full_name']} ⭐{r['stargazers_count']}: {r['description']}"
for r in repos
])
@mcp.tool()
async def create_issue(repo: str, title: str, body: str) -> str:
"""在指定仓库创建 Issue
Args:
repo: 仓库全名，如 owner/repo
title: Issue 标题
body: Issue 内容
"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"https://api.github.com/repos/{repo}/issues",
json={"title": title, "body": body},
headers={"Authorization": f"token {GITHUB_TOKEN}"}
)
if resp.status_code == 201:
return f"Issue 创建成功: {resp.json()['html_url']}"
return f"创建失败: {resp.text}"
# ============ Resources：可读取的数据 ============
@mcp.resource("github://repos/{owner}/{repo}/readme")
async def get_readme(owner: str, repo: str) -> str:
"""获取仓库的 README 内容"""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.github.com/repos/{owner}/{repo}/readme",
headers={
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.raw"
}
)
return resp.text if resp.status_code == 200 else "README 未找到"
@mcp.resource("github://user/starred")
async def get_starred() -> str:
"""获取用户 Star 的仓库列表"""
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.github.com/user/starred",
headers={"Authorization": f"token {GITHUB_TOKEN}"}
)
repos = resp.json()[:10]
return "\n".join([f"- {r['full_name']}" for r in repos])
# ============ Prompts：预定义模板 ============
@mcp.prompt()
def code_review_prompt(repo: str, pr_number: int) -> str:
"""生成代码审查的提示词模板"""
return f"""请审查 {repo} 仓库的 PR #{pr_number}。
审查要点：
1. 代码逻辑是否正确
2. 是否有潜在的性能问题
3. 代码风格是否符合规范
4. 是否有足够的测试覆盖
请先获取 PR 详情，然后逐一分析。"""
@mcp.prompt()
def issue_template(title: str) -> str:
"""生成 Issue 创建模板"""
return f"""请创建一个规范的 GitHub Issue。
标题: {title}
请按以下格式填写内容：
## 问题描述
[简述问题]
## 复现步骤
1.
2.
## 期望行为
[描述期望的正确行为]
## 环境信息
- 操作系统:
- 版本号:"""
# 运行服务
if __name__ == "__main__":
mcp.run()pip install mcp httpx{
"mcpServers": {
"github": {
"command": "python",
"args": ["/path/to/github_server.py"],
"env": {
"GITHUB_TOKEN": "your_token_here"
}
}
}
}用户: 帮我搜索 Python 的 RAG 相关项目
助手: [调用 search_repos(query="RAG", language="python")]
找到以下项目：
- langchain-ai/langchain ⭐75000: LLM 应用开发框架
- run-llama/llama_index ⭐30000: 数据框架连接 LLM
...