🚀 使用 Python 和千帆平台构建位置服务代理机器人
本项目借助 Python 和千帆平台，助力你快速构建位置服务代理机器人，为你提供便捷的位置相关服务。
🚀 快速开始
按照以下步骤，你可以快速搭建并运行位置服务代理机器人。
📦 安装指南
安装依赖项
要运行这个项目，您需要先安装以下依赖项：
pip install -r requirements.txt
配置 MCP Server
创建一个 config.py 文件，并添加以下内容：
class MCPConfig:
def __init__(self):
self.app_id = "your_app_id" # 替换为您的千帆应用 ID
self.token = "your_token" # 替换为您的千帆令牌
下载和安装千帆 SDK
您可以从 千帆官方文档 下载并安装 SDK。
创建 Agent 应用
在千帆平台上创建一个新的应用，并获取 app_id 和 token。
配置 MCP Server 环境变量
将 MCP Server 的环境变量设置为您的应用 ID 和令牌：
export APPBUILDER_TOKEN="your_token"
启动 MCP Server
运行以下命令启动 MCP Server：
python main.py
💻 使用示例
基础用法
将以下代码保存为 app.py，并按需修改路径和参数：
import os
import asyncio
import appbuilder
from appbuilder.core.console.appbuilder_client.async_event_handler import (
AsyncAppBuilderEventHandler,
)
from appbuilder.mcp_server.client import MCPClient
class MyEventHandler(AsyncAppBuilderEventHandler):
def __init__(self, mcp_client):
super().__init__()
self.mcp_client = mcp_client
def get_current_weather(self, location=None, unit="摄氏度"):
return "{} 的温度是 {} {}".format(location, 20, unit)
async def interrupt(self, run_context, run_response):
thought = run_context.current_thought
# 绿色打印
print("\033[1;31m", "-> Agent 中间思考: ", thought, "\033[0m")
tool_output = []
for tool_call in run_context.current_tool_calls:
tool_res = ""
if tool_call.function.name == "get_current_weather":
tool_res = self.get_current_weather(**tool_call.function.arguments)
else:
print(
"\033[1;32m",
"MCP工具名称: {}, MCP参数:{}\n".format(tool_call.function.name, tool_call.function.arguments),
"\033[0m",
)
mcp_server_result = await self.mcp_client.call_tool(
tool_call.function.name, tool_call.function.arguments
)
print("\033[1;33m", "MCP结果: {}\n\033[0m".format(mcp_server_result))
for i, content in enumerate(mcp_server_result.content):
if content.type == "text":
tool_res += mcp_server_result.content[i].text
tool_output.append(
{
"tool_call_id": tool_call.id,
"tool_response": tool_res
}
)
return tool_output
async def main():
# 初始化 MCP 客户端
mcp_client = MCPClient()
await mcp_client.start()
# 创建EventHandler并连接到MCP客户端
event_handler = MyEventHandler(mcp_client)
await mcp_client.register_event_handler(event_handler)
# 等待用户输入
while True:
query = input("请输入查询: ")
response = await mcp_client.process_query(query)
print("\033[1;34m", "Agent 响应:", response, "\033[0m")
if __name__ == "__main__":
asyncio.run(main())
运行机器人
运行 app.py，您将看到一个交互式界面，可以输入位置查询。
示例调用
# 启动 MCP Server
python main.py
# 在另一个终端中启动 Agent
python app.py
输入如“北京天气如何？”或“附近有哪些酒店？”，机器人将处理并返回结果。
✨ 主要特性
- 自定义地理位置服务：您可以根据需要扩展
MCPConfig类，添加更多位置相关服务。 - 多语言支持：通过调整日志和提示信息，支持多种语言。
- 高级分析：集成更多数据分析工具，提供更详细的报告。
📄 许可证
此项目遵循 MIT 协议。请在使用前仔细阅读许可条款。