将智能体接入真实 API 与数据库
构建能跟 REST API、数据库与文件系统交互的智能体。
🎯 你将学到
- 理解并应用本课涵盖的核心概念
超越 Mock 工具
每个智能体教程都展示返回假数据的 mock 函数。「这是你的股价：$100。」真正的挑战——也是大多数生产系统花时间的地方——是把真实 API 包装成可靠的智能体工具。
真实 API 有认证、限流、分页、不一致的响应格式与瞬时故障。你的智能体工具需要优雅地处理所有这些，因为智能体没有 HTTP 头或重试逻辑的概念。它只期望问一个问题、得到一个答案。
本教程构建一个 GitHub issue 分流（triage）智能体——一个实用工具，能搜索开放 issue、给它们打标签、把它们分配给团队成员并发布分流评论。每一步都涉及真实的 API 调用。
GitHub Issue 分流智能体
用例： 一个开发团队有 200 个开放的 GitHub issue。该智能体将：
- 抓取所有开放 issue，并识别出未打标签的
- 分析每个 issue 的内容，并建议合适的标签
- 基于分析应用标签
- 基于专业领域把 issue 分配给正确的团队成员
- 发布标准化的分流评论
这同时需要读与写操作——一个真实的生产场景。
把 REST API 包装成智能体工具
任何 REST API 工具的模式都一样：
import requests
import os
import time
from typing import Optional
from functools import wraps
# GitHub API configuration
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
GITHUB_API_BASE = "https://api.github.com"
HEADERS = {
"Authorization": f"Bearer {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json",
"X-GitHub-Api-Version": "2022-11-28"
}
def github_request(
method: str,
endpoint: str,
params: dict = None,
json_body: dict = None,
max_retries: int = 3
) -> dict:
"""
Make a GitHub API request with retry logic and rate limit handling.
This is the foundation all GitHub tools are built on.
"""
url = f"{GITHUB_API_BASE}{endpoint}"
for attempt in range(max_retries):
try:
response = requests.request(
method=method,
url=url,
headers=HEADERS,
params=params,
json=json_body,
timeout=15
)
# Handle rate limiting
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate limited. Waiting {retry_after} seconds...")
time.sleep(retry_after)
continue
# Handle 403 with rate limit headers (GitHub-specific)
if response.status_code == 403:
remaining = int(response.headers.get("X-RateLimit-Remaining", 1))
if remaining == 0:
reset_time = int(response.headers.get("X-RateLimit-Reset", time.time() + 60))
wait_time = max(0, reset_time - time.time()) + 1
print(f"Rate limit hit. Waiting {wait_time:.0f} seconds...")
time.sleep(wait_time)
continue
# Raise for other 4xx/5xx errors
response.raise_for_status()
# Handle empty responses (e.g., 204 No Content)
if response.status_code == 204:
return {"success": True}
return response.json()
except requests.exceptions.Timeout:
if attempt == max_retries - 1:
return {"error": f"Request timed out after {max_retries} attempts"}
time.sleep(2 ** attempt) # Exponential backoff
except requests.exceptions.HTTPError as e:
return {"error": f"HTTP {response.status_code}: {str(e)}",
"body": response.text[:500]}
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
return {"error": f"Request failed: {str(e)}"}
time.sleep(2 ** attempt)
return {"error": "All retry attempts failed"}
这个 github_request 函数处理了三个常见的现实问题：
- 认证：token 通过 header 加到每个请求上
- 限流：检测 429 响应并在重试前等待
- 瞬时故障：在超时/错误时用指数退避重试
你在它之上构建的每个工具都会自动继承这些属性。
定义 GitHub 工具
import json
def list_open_issues(repo: str, labels: str = "", per_page: int = 30) -> str:
"""
Fetch open issues from a repository.
Args:
repo: Repository in format "owner/repo" (e.g. "pytorch/pytorch")
labels: Comma-separated labels to filter by (optional)
per_page: Number of issues to fetch (max 100)
"""
params = {
"state": "open",
"per_page": min(per_page, 100),
"sort": "created",
"direction": "desc"
}
if labels:
params["labels"] = labels
result = github_request("GET", f"/repos/{repo}/issues", params=params)
if "error" in result:
return f"Error fetching issues: {result['error']}"
# Format for the agent — return clean, relevant fields
issues = []
for issue in result:
if "pull_request" in issue: # Skip PRs, only real issues
continue
issues.append({
"number": issue["number"],
"title": issue["title"],
"body": issue.get("body", "")[:500], # Truncate long bodies
"labels": [l["name"] for l in issue.get("labels", [])],
"assignees": [a["login"] for a in issue.get("assignees", [])],
"created_at": issue["created_at"][:10]
})
return json.dumps(issues, indent=2)
def apply_label_to_issue(repo: str, issue_number: int, labels: list[str]) -> str:
"""
Apply labels to a GitHub issue.
Args:
repo: Repository in format "owner/repo"
issue_number: The issue number (integer)
labels: List of label names to apply (must already exist in the repo)
"""
result = github_request(
"POST",
f"/repos/{repo}/issues/{issue_number}/labels",
json_body={"labels": labels}
)
if "error" in result:
return f"Error applying labels: {result['error']}"
applied = [l["name"] for l in result]
return f"Applied labels {applied} to issue #{issue_number}"
def assign_issue(repo: str, issue_number: int, assignees: list[str]) -> str:
"""
Assign a GitHub issue to one or more team members.
Args:
repo: Repository in format "owner/repo"
issue_number: The issue number (integer)
assignees: List of GitHub usernames to assign
"""
result = github_request(
"POST",
f"/repos/{repo}/issues/{issue_number}/assignees",
json_body={"assignees": assignees}
)
if "error" in result:
return f"Error assigning issue: {result['error']}"
return f"Assigned issue #{issue_number} to {assignees}"
def post_comment(repo: str, issue_number: int, body: str) -> str:
"""
Post a comment on a GitHub issue.
Args:
repo: Repository in format "owner/repo"
issue_number: The issue number (integer)
body: The comment text (markdown supported)
"""
result = github_request(
"POST",
f"/repos/{repo}/issues/{issue_number}/comments",
json_body={"body": body}
)
if "error" in result:
return f"Error posting comment: {result['error']}"
return f"Comment posted on issue #{issue_number}: {result.get('html_url', '')}"
构建分流智能体
import anthropic
client = anthropic.Anthropic()
TRIAGE_TOOLS = [
{
"name": "list_open_issues",
"description": """Fetch open GitHub issues from a repository.
Use this first to see what issues need triage.
Returns: list of issues with number, title, body preview, current labels, and assignees.""",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string", "description": "Repository as 'owner/repo'"},
"labels": {"type": "string", "description": "Filter by label (optional)"},
"per_page": {"type": "integer", "description": "How many issues to fetch (default 30)"}
},
"required": ["repo"]
}
},
{
"name": "apply_label_to_issue",
"description": """Apply labels to a GitHub issue.
Use after analyzing an issue's content to categorize it.
Common labels: 'bug', 'enhancement', 'documentation', 'question', 'good first issue'.""",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"issue_number": {"type": "integer"},
"labels": {"type": "array", "items": {"type": "string"}}
},
"required": ["repo", "issue_number", "labels"]
}
},
{
"name": "assign_issue",
"description": """Assign a GitHub issue to team members.
Use based on the issue's technical area and team expertise.""",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"issue_number": {"type": "integer"},
"assignees": {"type": "array", "items": {"type": "string"}}
},
"required": ["repo", "issue_number", "assignees"]
}
},
{
"name": "post_comment",
"description": """Post a triage comment on a GitHub issue.
Use to acknowledge the issue and provide initial guidance.""",
"input_schema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"issue_number": {"type": "integer"},
"body": {"type": "string", "description": "Markdown comment body"}
},
"required": ["repo", "issue_number", "body"]
}
}
]
TOOL_FUNCTIONS = {
"list_open_issues": list_open_issues,
"apply_label_to_issue": apply_label_to_issue,
"assign_issue": assign_issue,
"post_comment": post_comment
}
TRIAGE_SYSTEM = """You are a GitHub issue triage agent. Your job is to help development teams
manage their issue backlog.
Team expertise:
- @alice-dev: backend/API, Python, databases
- @bob-frontend: frontend, JavaScript/React, CSS
- @carol-infra: DevOps, Docker, Kubernetes, CI/CD
- @dave-ml: machine learning, model training, data pipelines
When triaging issues:
1. Fetch unlabeled issues first
2. Analyze title and body to determine: type (bug/enhancement/question/docs), area (backend/frontend/infra/ml)
3. Apply appropriate labels
4. Assign to the relevant team member based on area
5. Post a brief, friendly triage comment acknowledging the issue
Be efficient — triage multiple issues in one pass before posting comments."""
def run_triage_agent(repo: str, max_issues: int = 10) -> str:
messages = [{
"role": "user",
"content": f"Triage unlabeled open issues in {repo}. Process up to {max_issues} issues."
}]
for _ in range(20): # More iterations for batch operations
response = client.messages.create(
model="claude-opus-4-5",
max_tokens=4096,
system=TRIAGE_SYSTEM,
tools=TRIAGE_TOOLS,
messages=messages
)
if response.stop_reason == "end_turn":
return response.content[0].text
if response.stop_reason == "tool_use":
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
fn = TOOL_FUNCTIONS[block.name]
result = fn(**block.input)
print(f"[{block.name}] {str(block.input)[:80]}...")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "user", "content": tool_results})
return "Triage complete (max iterations reached)."
# Run
summary = run_triage_agent("your-org/your-repo", max_issues=5)
print(summary)
REST vs 数据库 vs 文件系统工具
不同类型的外部系统有不同的模式：
REST API 工具（如上面的 GitHub）：
- 通过 header 认证（API key、OAuth token）
- 响应是 JSON——始终校验结构
- 限流很常见——要内建重试逻辑
- 分页意味着大数据集你可能需要多次调用
数据库工具：
- 使用参数化查询防止 SQL 注入
- 限制结果大小——一个没有 LIMIT 的
SELECT *可能返回数百万行 - 尽可能把写操作包在事务里
- 返回计数而非原始数据用于校验
import sqlite3
def query_database(sql: str, params: tuple = ()) -> str:
"""Execute a read-only SQL query. Max 100 rows returned."""
# Safety: only allow SELECT statements
if not sql.strip().upper().startswith("SELECT"):
return "Error: only SELECT queries are allowed"
conn = sqlite3.connect("app.db")
cursor = conn.cursor()
cursor.execute(sql + " LIMIT 100", params) # Always add LIMIT
columns = [d[0] for d in cursor.description]
rows = cursor.fetchall()
conn.close()
return json.dumps({"columns": columns, "rows": rows, "count": len(rows)})
文件系统工具：
- 始终校验路径，防止目录穿越攻击
- 设定工作目录，并拒绝其之外的路径
- 返回带大小限制的文件内容
- 记录所有写操作
import os
SAFE_DIR = "/app/workspace" # Only allow access within this directory
def read_file_safe(filepath: str) -> str:
"""Read a file, restricted to the workspace directory."""
# Resolve the full path and check it's within SAFE_DIR
full_path = os.path.realpath(os.path.join(SAFE_DIR, filepath))
if not full_path.startswith(SAFE_DIR):
return "Error: Access denied. Path is outside the workspace."
if not os.path.exists(full_path):
return f"Error: File not found: {filepath}"
file_size = os.path.getsize(full_path)
if file_size > 100_000: # 100KB limit
return f"Error: File too large ({file_size} bytes). Max 100KB."
with open(full_path) as f:
return f.read()
总结
- 把所有 REST API 调用包在一个共享的请求函数里，处理认证、限流与重试
- 把工具输出格式化成干净、对智能体友好的——而非原始 API 响应
- REST 工具需要认证 header 和重试逻辑；数据库工具需要参数化查询和 LIMIT 子句；文件系统工具需要路径校验
- GitHub 分流智能体展示了一个完整的可写智能体：读 issue、打标签、分配并评论
- 从工具返回结构化的错误消息——智能体能就错误进行推理并尝试替代方案
- 在把工具接入智能体循环前，始终先在隔离环境针对真实 API 测试工具
#AI#tutorial#superml#agent#api#database#github#integration