模型上下文协议（MCP）允许人工智能代理通过标准化接口调用外部工具。多工具 MCP 代理可以将 Web 搜索、文件系统访问、数据库查询和自定义 API 组合在单个对话中。本教程使用 Scavio API 支持的搜索工具、文件工具和计算器工具构建 MCP 代理。该模式可扩展到任意数量的工具。通过 Scavio 进行的搜索呼叫每次费用为 0.005 美元。
前置条件
- 已安装 Python 3.10+
- 安装 mcp 软件包（pip install mcp）
- 来自 scavio.dev 的 Scavio API 密钥
- 对工具使用模式的基本了解
操作指南
步骤 1: 定义 MCP 搜索工具服务器
创建一个公开 web_search 函数的 MCP 工具服务器。该服务器处理 API 调用并将结构化结果返回给代理。
Python
from mcp.server import Server
from mcp.types import Tool, TextContent
import requests, os, json
server = Server('search-tools')
API_KEY = os.environ['SCAVIO_API_KEY']
@server.tool()
async def web_search(query: str, country: str = 'us') -> str:
"""Search the web for current information."""
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
results = resp.json().get('organic_results', [])[:5]
return json.dumps([{'title': r['title'], 'url': r['link'],
'snippet': r.get('snippet', '')} for r in results], indent=2)步骤 2: 在同一服务器上添加TikTok搜索工具
添加第二个搜索 TikTok 的工具。这两种工具共享同一个 MCP 服务器，因此代理可以在单个会话中使用任一工具。
Python
@server.tool()
async def tiktok_search(keyword: str, count: int = 10) -> str:
"""Search TikTok for videos about a topic."""
resp = requests.post('https://api.scavio.dev/api/v1/tiktok/search/videos',
headers={'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'},
json={'keyword': keyword, 'count': count, 'cursor': 0})
videos = resp.json().get('data', {}).get('videos', [])
return json.dumps([{'author': v.get('author', {}).get('nickname', ''),
'desc': v.get('desc', ''), 'plays': v.get('stats', {}).get('playCount', 0)}
for v in videos], indent=2)步骤 3: 添加计算器和文件阅读器工具
通过添加非搜索工具来演示多工具模式。代理根据用户查询决定使用哪个工具。
Python
import math
@server.tool()
async def calculate(expression: str) -> str:
"""Evaluate a mathematical expression safely."""
allowed = set('0123456789+-*/.() ')
if not all(c in allowed for c in expression):
return 'Error: only numeric expressions allowed'
try:
result = eval(expression)
return str(result)
except Exception as e:
return f'Error: {e}'
@server.tool()
async def read_file(path: str) -> str:
"""Read a local text file."""
try:
with open(path, 'r') as f:
content = f.read(10000) # limit to 10KB
return content
except FileNotFoundError:
return f'File not found: {path}'步骤 4: 配置 MCP 服务器清单
创建 .mcp.json 配置文件，告诉 MCP 客户端在哪里可以找到您的工具服务器以及哪些工具可用。
JSON
// .mcp.json
{
"mcpServers": {
"search-tools": {
"command": "python",
"args": ["search_tools_server.py"],
"env": {
"SCAVIO_API_KEY": "your_scavio_api_key"
}
}
}
}
// The server exposes: web_search, tiktok_search, calculate, read_file步骤 5: 运行 MCP 服务器
使用 stdio 传输启动服务器。任何与 MCP 兼容的客户端（Claude Desktop、Cursor、自定义代理）现在都可以连接并使用所有四种工具。
Python
import asyncio
from mcp.server.stdio import stdio_server
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream)
if __name__ == '__main__':
asyncio.run(main())
# Test manually:
# echo '{"jsonrpc":"2.0","method":"tools/list","id":1}' | python search_tools_server.py
# Should list: web_search, tiktok_search, calculate, read_filePython 示例
Python
from mcp.server import Server
from mcp.server.stdio import stdio_server
import requests, os, json, asyncio
server = Server('search-tools')
API_KEY = os.environ['SCAVIO_API_KEY']
@server.tool()
async def web_search(query: str, country: str = 'us') -> str:
"""Search the web for current information."""
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
results = resp.json().get('organic_results', [])[:5]
return json.dumps([{'title': r['title'], 'snippet': r.get('snippet', '')} for r in results])
@server.tool()
async def tiktok_search(keyword: str, count: int = 10) -> str:
"""Search TikTok videos."""
resp = requests.post('https://api.scavio.dev/api/v1/tiktok/search/videos',
headers={'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'application/json'},
json={'keyword': keyword, 'count': count, 'cursor': 0})
return json.dumps(resp.json().get('data', {}).get('videos', [])[:5])
async def main():
async with stdio_server() as (read, write):
await server.run(read, write)
if __name__ == '__main__':
asyncio.run(main())JavaScript 示例
JavaScript
// MCP servers are typically Python; this shows a JS HTTP wrapper
const API_KEY = process.env.SCAVIO_API_KEY;
const tools = {
async web_search({ query, country = 'us' }) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: country })
});
const data = await resp.json();
return (data.organic_results || []).slice(0, 5)
.map(r => ({ title: r.title, snippet: r.snippet || '' }));
},
async tiktok_search({ keyword, count = 10 }) {
const resp = await fetch('https://api.scavio.dev/api/v1/tiktok/search/videos', {
method: 'POST',
headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' },
body: JSON.stringify({ keyword, count, cursor: 0 })
});
return (await resp.json()).data?.videos || [];
}
};
async function main() {
const results = await tools.web_search({ query: 'MCP protocol 2026' });
console.log(JSON.stringify(results, null, 2));
}
main().catch(console.error);预期输出
JSON
Available tools:
- web_search: Search the web for current information
- tiktok_search: Search TikTok for videos about a topic
- calculate: Evaluate a mathematical expression
- read_file: Read a local text file
[
{"title": "MCP Protocol Documentation", "snippet": "The Model Context Protocol..."}
{"title": "Building MCP Agents in 2026", "snippet": "A guide to creating..."}
]
Cost: $0.005 per web_search call, $0.005 per tiktok_search call