天地图

🚀 使用 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 協議。請在使用前仔細閱讀許可條款。