import osfrom openai import OpenAI

client = OpenAI(
     api_key="Your-sdk",  # 替换为你的 AI Studio 访问令牌
     base_url="https://aistudio.baidu.com/llm/lmapi/v3",  # AI Studio 大模型 API 服务域名)# 示例：简单对话# chat_completion = client.chat.completions.create(#     messages=[#         {'role': 'system', 'content': '你是 AI Studio 实训AI开发平台的开发者助理，你精通开发相关的知识，负责给开发者提供搜索帮助建议。'},#         {'role': 'user', 'content': '你好，请介绍一下AI Studio'}#     ],#     model="ernie-3.5-8k", # 或者 ernie-4.0-8k 等其他适用模型# )# print(chat_completion.choices[0].message.content)

# ernie_client.py (或Notebook初始化块)import osfrom openai import OpenAI# AI Studio Ernie API 配置# 请确保使用你自己的有效API Key和正确的Base URLASTUDIO_API_KEY = os.getenv("AISTUDIO_API_KEY", "") # 从环境变量读取或直接填写ASTUDIO_BASE_URL = os.getenv("AISTUDIO_BASE_URL", "https://aistudio.baidu.com/llm/lmapi/v3")

client = OpenAI(
    api_key=ASTUDIO_API_KEY,
    base_url=ASTUDIO_BASE_URL,
)def get_ernie_response(user_prompt, system_prompt="你是一个乐于助人的AI助手。", model="ernie-3.5-8k"):
    """调用Ernie API获取回复"""
    try:
        chat_completion = client.chat.completions.create(
            messages=[
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': user_prompt}
            ],
            model=model,
        )        return chat_completion.choices[0].message.content    except Exception as e:        print(f"调用Ernie API失败: {e}")        return f"无法获取Ernie的回应，错误: {e}" # 返回错误信息，而不是None# 测试一下 (可选)# if __name__ == "__main__":#     response = get_ernie_response("你好，请介绍一下北京的天气如何？")#     print(response)

# weather_agent_server.pyimport timefrom flask import Flask, request, jsonifyfrom python_a2a import A2AServer, AgentCard, AgentSkill, Task, TaskStatus, TaskState, Message, TextContent, MessageRoleclass WeatherAgent(A2AServer):
    def __init__(self, port=5001):
        self.port = port
        agent_card = AgentCard(
            name="Weather Agent",
            description="提供指定城市的天气信息",
            url=f"http://localhost:{self.port}",
            version="1.0.0",
            skills=[
                AgentSkill(
                    name="GetCityWeather",
                    description="获取一个城市的当前天气",
                    tags=["weather", "location", "forecast"]
                )
            ]
        )        super().__init__(agent_card=agent_card)    def _get_simulated_weather(self, city):
        print(f"[WeatherAgent] 正在为城市 '{city}' 查询模拟天气...")
        time.sleep(1) # 模拟网络延迟
        weather_conditions = {            "北京": "晴朗，25°C，微风。",            "上海": "多云转小雨，22°C，东南风3级。",            "广州": "雷阵雨，28°C，注意防范。",            "深圳": "晴间多云，29°C，空气质量良。",            "伦敦": "持续小雨，15°C，请带好雨具。",            "巴黎": "阳光明媚，23°C，适合户外活动。"
        }        return weather_conditions.get(city, f"抱歉，暂时无法获取'{city}'的天气信息。也许可以问问Ernie？")    # 如何集成Ernie (示例性注释，实际使用时需要取消注释并确保 ernie_client.py 可导入)
    # from ernie_client import get_ernie_response
    # def _get_weather_from_ernie(self, city):
    #     print(f"[WeatherAgent] 正在通过Ernie查询城市 '{city}' 的天气...\")
    #     prompt = f"请告诉我{city}今天的天气怎么样？请简要描述。\"
    #     weather_info = get_ernie_response(prompt, system_prompt="你是一个天气预报助手。")
    #     return weather_info

    def handle_task(self, task: Task) -> Task:
        query_text = ""
        if task.message and task.message.content and hasattr(task.message.content, 'text'):
            query_text = task.message.content.text
        
        city = query_text # 假设查询文本直接是城市名
        
        # **核心逻辑：获取天气**
        # 默认使用模拟数据
        weather_data = self._get_simulated_weather(city)        # # 如果想使用Ernie获取天气 (取消下面一行的注释，并注释掉上面一行):
        # weather_data = self._get_weather_from_ernie(city)

        print(f"[WeatherAgent] 为 '{city}' 获取到的天气: {weather_data}")

        task.artifacts = [{            "parts": [{"type": "text", "text": weather_data}]
        }]
        task.status = TaskStatus(state=TaskState.COMPLETED)        return task# ---- Flask App Setup ----def create_app(agent_instance):
    app = Flask(__name__)    @app.route('/agent.json', methods=['GET'])
    def get_agent_card():
        return jsonify(agent_instance.agent_card.to_dict())    @app.route('/a2a/agent.json', methods=['GET'])
    def get_a2a_agent_card(): # A2A标准兼容端点
        return jsonify(agent_instance.agent_card.to_dict())    @app.route('/tasks/send', methods=['POST'])
    def handle_send_task():
        try:
            task_data = request.json            # 兼容JSON-RPC风格的请求
            if "jsonrpc" in task_data and "method" in task_data and task_data["method"] == "tasks/send":
                task_dict = task_data.get("params", {})            else:
                task_dict = task_data
            
            task = Task.from_dict(task_dict)
            updated_task = agent_instance.handle_task(task)
            response_data = updated_task.to_dict()            if "jsonrpc" in task_data:                 return jsonify({"jsonrpc": "2.0", "id": task_data.get("id"), "result": response_data})            return jsonify(response_data)        except Exception as e:
            error_response = {"code": -32603, "message": str(e)}            if "jsonrpc" in request.json:                return jsonify({"jsonrpc": "2.0", "id": request.json.get("id"), "error": error_response}), 500
            return jsonify({"error": error_response}), 500
            
    # 兼容旧版Message端点 (可选，推荐使用Task)    @app.route('/', methods=['POST'])
    def handle_root_message():
        try:
            data = request.json            # 尝试将整个请求体作为Task处理
            if isinstance(data, dict) and "id" in data and "status" in data: # 看起来像Task
                 return handle_send_task()            # 否则，尝试作为Message处理 (简化版)
            text_content = data.get("content", {}).get("text", "") if isinstance(data.get("content"), dict) else str(data)
            message = Message(content=TextContent(text=text_content), role=MessageRole.USER)            # 模拟任务转换
            task = Task(id=f"task-{time.time_ns()}", message=message, status=TaskStatus(state=TaskState.PENDING))
            updated_task = agent_instance.handle_task(task)            # 从Task结果中提取文本给Message响应
            response_text = "处理完成。"
            if updated_task.artifacts and updated_task.artifacts[0].get("parts"):
                response_text = updated_task.artifacts[0]["parts"][0].get("text", response_text)
            
            response_message = Message(
                content=TextContent(text=response_text),
                role=MessageRole.AGENT,
                message_id=f"response-{time.time_ns()}",
                parent_message_id=data.get("message_id")
            )            return jsonify(response_message.to_dict())        except Exception as e:            return jsonify({"error": str(e)}), 400
            
    return appif __name__ == "__main__":
    PORT = 5001
    weather_agent = WeatherAgent(port=PORT)
    app = create_app(weather_agent)    print(f"WeatherAgent 正在启动，监听端口: http://localhost:{PORT}")
    app.run(host='0.0.0.0', port=PORT, debug=False)

python weather_agent_server.py

# activity_agent_server.pyimport timeimport re # 用于从查询中提取天气信息from flask import Flask, request, jsonifyfrom python_a2a import A2AServer, AgentCard, AgentSkill, Task, TaskStatus, TaskState, Message, TextContent, MessageRoleclass ActivityAgent(A2AServer):
    def __init__(self, port=5002):
        self.port = port
        agent_card = AgentCard(
            name="Activity Agent",
            description="根据城市和天气推荐活动",
            url=f"http://localhost:{self.port}",
            version="1.0.0",
            skills=[
                AgentSkill(
                    name="GetActivityRecommendations",
                    description="获取基于天气和地点的活动推荐",
                    tags=["activity", "recommendation", "travel", "leisure"]
                )
            ]
        )        super().__init__(agent_card=agent_card)    def _get_simulated_activities(self, city, weather_condition):
        print(f"[ActivityAgent] 正在为 '{city}' (天气: {weather_condition}) 生成模拟活动推荐...")
        time.sleep(1.5) # 模拟处理时间
        is_rainy_or_bad_weather = any(kw in weather_condition.lower() for kw in ["雨", "雷", "雪", "糟", "霾"])
        
        recommendations = {            "北京": {                "indoor": "故宫博物院深度游、国家博物馆看展、798艺术区感受艺术氛围。",                "outdoor": "颐和园漫步、长城徒步（天气好时）、什刹海胡同骑行。"
            },            "上海": {                "indoor": "上海博物馆探索历史、中华艺术宫赏析艺术、体验密室逃脱。",                "outdoor": "外滩漫步欣赏万 国建筑群、南京路步行街购物、豫园游览。"
            },            "广州": {                "indoor": "广东省博物馆、广州塔室内观光层、正佳广场极地海洋世界。",                "outdoor": "白云山登高望远、越秀公园五羊石像、珠江夜游。"
            },            "深圳": {                "indoor": "深圳博物馆、当代艺术与城市规划馆、万象城购物体验。",                "outdoor": "世界之窗（天气好时）、莲花山公园放风筝、深圳湾公园骑行。"
            },            "伦敦": {                "indoor": "大英博物馆、国家美术馆、自然历史博物馆。",                "outdoor": "海德公园野餐、伦敦眼俯瞰城市、泰晤士河畔散步。"
            },            "巴黎": {                "indoor": "卢浮宫、奥赛博物馆、蓬皮杜艺术中心。",                "outdoor": "埃菲尔铁塔下草坪休闲、塞纳河畔漫步、蒙马特高地探索。"
            }
        }
        
        city_activities = recommendations.get(city, {            "indoor": f"在{city}可以逛逛当地的博物馆或购物中心。",            "outdoor": f"在{city}可以探索当地的公园或著名地标。"
        })        
        if is_rainy_or_bad_weather:            return f"由于{city}的天气是'{weather_condition}'，推荐室内活动：{city_activities['indoor']}"
        else:            return f"{city}天气'{weather_condition}'，很棒！推荐户外活动：{city_activities['outdoor']}"

    # 如何集成Ernie (示例性注释)
    # from ernie_client import get_ernie_response
    # def _get_activities_from_ernie(self, city, weather_condition):
    #     print(f"[ActivityAgent] 正在通过Ernie为 '{city}' (天气: {weather_condition}) 生成活动推荐...\")
    #     prompt = f"我现在在{city}，天气是{weather_condition}。请为我推荐一些合适的活动，户外和室内都可以考虑，并简要说明理由。\"
    #     activities = get_ernie_response(prompt, system_prompt="你是一个旅行活动推荐助手。")
    #     return activities

    def _extract_info_from_query(self, query_text):
        # 这是一个非常简化的提取逻辑，实际应用中可能需要更复杂的NLP
        # 假设查询格式是 "为[城市]在[天气状况]下推荐活动"
        # 或者直接从上下文中获取城市和天气
        city = "未知城市"
        weather = "未知天气"

        # 尝试从文本中提取城市，例如北京、上海等
        known_cities = ["北京", "上海", "广州", "深圳", "伦敦", "巴黎"]        for c in known_cities:            if c in query_text:
                city = c                break
        
        # 尝试提取天气信息，例如包含"晴朗"、"雨"等关键词
        # 这里我们假设天气信息在查询文本的后半部分，或者作为独立参数传入
        # 在我们的工作流中，天气信息会由WeatherAgent提供，并作为输入的一部分
        # 所以这里的提取逻辑更多是针对直接调用此Agent的情况
        weather_keywords = {            "晴朗": ["晴", "太阳", "sunny"],            "多云": ["多云", "cloudy"],            "阴": ["阴天", "overcast"],            "雨": ["雨", "rainy", "雷阵雨"],            "雪": ["雪", "snowy"]
        }        for condition, kws in weather_keywords.items():            if any(kw in query_text for kw in kws):
                weather = condition                break
        # 如果没有通过关键词匹配到，则尝试取查询文本中关于天气的描述
        # 比如，如果query_text是 "北京的天气是晴朗，25°C，微风。" 中的后半段
        # 这里的逻辑需要根据实际的输入格式来完善
        if weather == "未知天气": #如果上面没匹配到，尝试更通用的提取
            match = re.search(r"天气是([^，。]+)|状况是([^，。]+)", query_text)            if match:
                weather = match.group(1) or match.group(2) or "宜人"
            elif "天气" in query_text: # 退化情况，如果只说了天气，但没描述
                 weather = "宜人（具体情况未知）"

        return city, weather    def handle_task(self, task: Task) -> Task:
        query_text = ""
        if task.message and task.message.content and hasattr(task.message.content, 'text'):
            query_text = task.message.content.text # 例如 "为北京在晴朗，25°C，微风。下推荐活动"
        
        # 从查询文本中解析出城市和天气状况
        # 在我们的工作流中，城市是初始输入，天气是上一个Agent的输出
        # query_text的格式会是 Flow 构建的，例如 "Recommend activities in Beijing given weather: 晴朗，25°C，微风。"
        
        city_match = re.search(r"in (\w+)", query_text, re.IGNORECASE)
        city = city_match.group(1) if city_match else "北京" # 默认北京
        
        weather_match = re.search(r"weather: (.+)", query_text, re.IGNORECASE)
        weather_condition = weather_match.group(1) if weather_match else "天气不错" # 默认天气不错
        if not weather_match: # 尝试从更简单的格式提取天气，如果上一个Agent直接返回天气描述
            if "推荐活动" in query_text: # 假设我们构建的查询是 "为[城市]在[天气描述]下推荐活动"
                 parts = query_text.split("在")                 if len(parts) > 1:
                    potential_weather = parts[-1].replace("下推荐活动","").strip()                    if potential_weather:
                        weather_condition = potential_weather        # **核心逻辑：获取活动推荐**
        # 默认使用模拟数据
        activity_data = self._get_simulated_activities(city, weather_condition)        # # 如果想使用Ernie获取活动 (取消下面一行的注释，并注释掉上面一行):
        # activity_data = self._get_activities_from_ernie(city, weather_condition)
        
        print(f"[ActivityAgent] 为 '{city}' (天气: {weather_condition}) 推荐的活动: {activity_data}")

        task.artifacts = [{            "parts": [{"type": "text", "text": activity_data}]
        }]
        task.status = TaskStatus(state=TaskState.COMPLETED)        return task# ---- Flask App Setup (与WeatherAgent类似) ----def create_app(agent_instance):
    app = Flask(__name__)    @app.route('/agent.json', methods=['GET'])
    def get_agent_card():
        return jsonify(agent_instance.agent_card.to_dict())    @app.route('/a2a/agent.json', methods=['GET'])
    def get_a2a_agent_card():
        return jsonify(agent_instance.agent_card.to_dict())    @app.route('/tasks/send', methods=['POST'])
    def handle_send_task():
        try:
            task_data = request.json            if "jsonrpc" in task_data and "method" in task_data and task_data["method"] == "tasks/send":
                task_dict = task_data.get("params", {})            else:
                task_dict = task_data
            
            task = Task.from_dict(task_dict)
            updated_task = agent_instance.handle_task(task)
            response_data = updated_task.to_dict()            if "jsonrpc" in task_data:                 return jsonify({"jsonrpc": "2.0", "id": task_data.get("id"), "result": response_data})            return jsonify(response_data)        except Exception as e:
            error_response = {"code": -32603, "message": str(e)}            if "jsonrpc" in request.json:                return jsonify({"jsonrpc": "2.0", "id": request.json.get("id"), "error": error_response}), 500
            return jsonify({"error": error_response}), 500
                @app.route('/', methods=['POST'])
    def handle_root_message(): # 兼容旧版
        try:
            data = request.json            if isinstance(data, dict) and "id" in data and "status" in data:                 return handle_send_task()
            text_content = data.get("content", {}).get("text", "") if isinstance(data.get("content"), dict) else str(data)
            message = Message(content=TextContent(text=text_content), role=MessageRole.USER)
            task = Task(id=f"task-{time.time_ns()}", message=message, status=TaskStatus(state=TaskState.PENDING))
            updated_task = agent_instance.handle_task(task)
            response_text = "处理完成。"
            if updated_task.artifacts and updated_task.artifacts[0].get("parts"):
                response_text = updated_task.artifacts[0]["parts"][0].get("text", response_text)
            response_message = Message(
                content=TextContent(text=response_text),
                role=MessageRole.AGENT,
                message_id=f"response-{time.time_ns()}",
                parent_message_id=data.get("message_id")
            )            return jsonify(response_message.to_dict())        except Exception as e:            return jsonify({"error": str(e)}), 400
            
    return appif __name__ == "__main__":
    PORT = 5002
    activity_agent = ActivityAgent(port=PORT)
    app = create_app(activity_agent)    print(f"ActivityAgent 正在启动，监听端口: http://localhost:{PORT}")
    app.run(host='0.0.0.0', port=PORT, debug=False)

python activity_agent_server.py

# 在Jupyter Notebook单元格中或 run_workflow.py 中执行import timefrom python_a2a import AgentNetwork, Flow, Message, TextContent, MessageRole,A2AClient# 0. (可选) 导入Ernie客户端，如果要在工作流本身或结果处理中使用from ernie_client import get_ernie_response# 1. 定义Agent网络# 这里的URL应与我们启动Agent服务时指定的端口一致WEATHER_AGENT_URL = "http://localhost:5001"ACTIVITY_AGENT_URL = "http://localhost:5002"# 确保WeatherAgent和ActivityAgent服务已在终端启动并正在运行！# 可以简单测试一下Agent是否可达 (可选)import requeststry:    print(f"尝试连接WeatherAgent: {WEATHER_AGENT_URL}/agent.json")
    weather_card = requests.get(f"{WEATHER_AGENT_URL}/agent.json", timeout=3).json()    print(f"WeatherAgent连接成功: {weather_card.get('name')}")    print(f"尝试连接ActivityAgent: {ACTIVITY_AGENT_URL}/agent.json")
    activity_card = requests.get(f"{ACTIVITY_AGENT_URL}/agent.json", timeout=3).json()    print(f"ActivityAgent连接成功: {activity_card.get('name')}")except requests.exceptions.ConnectionError as e:    print(f"连接Agent失败! 请确保 weather_agent_server.py 和 activity_agent_server.py 正在运行。错误: {e}")    # 如果连接失败，后续工作流会出错，可以根据情况决定是否中止
    # raise SystemExit("Agent服务未运行，工作流无法执行。") from eagent_network = AgentNetwork(name="Local Travel Assistant Network")# 直接使用URL添加Agent，AgentNetwork.add内部会创建A2AClient# 第二个参数名为 agent_or_url，可以直接传递URL字符串agent_network.add(name="weather_service", agent_or_url=WEATHER_AGENT_URL)
agent_network.add(name="activity_service", agent_or_url=ACTIVITY_AGENT_URL)print("\nAgent网络已配置:")for agent_info in agent_network.list_agents():    print(f"- {agent_info['name']}: {agent_info['url']}")# 2. 定义工作流# 用户希望查询的城市user_city_input = "北京"# user_city_input = "伦敦"# user_city_input = "火星" # 测试一下Agent的边界情况print(f"\n准备为城市 '{user_city_input}'规划行程。")# 使用Flow API定义工作流程# Flow的每一步都像是在构建一个请求链travel_flow = Flow(agent_network=agent_network, name="Smart Travel Planning Flow")# 第一步：调用 WeatherAgent 获取天气# .ask(agent_name_in_network, query_string_or_Task_Message_or_lambda)# 如果是字符串，则会被自动包装travel_flow.ask(
    agent_name="weather_service", 
    query=user_city_input
) 

# 第二步：条件分支，根据天气推荐活动travel_flow.if_contains("雨") # 如果天气结果包含"雨"# 如果包含"雨" (IF分支):# 2a. 使用 FunctionStep 构造特定于此分支的查询字符串travel_flow.execute_function(    lambda weather_report_from_step1: f"Recommend indoor activities in {user_city_input} given weather: {weather_report_from_step1}",    "{1}" # 参数：传递天气服务步骤的结果给 lambda)# 2b. 调用 ActivityAgent，使用 FunctionStep 生成的查询 (现在在 latest_result 中)travel_flow.ask(
    agent_name="activity_service",
    query="{latest_result}" # QueryStep 将从 context.data['latest_result'] 获取查询字符串)# 如果不包含"雨" (ELSE分支):travel_flow.else_branch()# 2c. 使用 FunctionStep 构造特定于此分支的查询字符串travel_flow.execute_function(    lambda weather_report_from_step1: f"Recommend outdoor activities in {user_city_input} given weather: {weather_report_from_step1}",    "{1}" # 参数：传递天气服务步骤的结果给 lambda)# 2d. 调用 ActivityAgent，使用 FunctionStep 生成的查询 (现在在 latest_result 中)travel_flow.ask(
    agent_name="activity_service",
    query="{latest_result}" # QueryStep 将从 context.data['latest_result'] 获取查询字符串)# 结束条件分支travel_flow.end_if()# 3. 执行工作流print("\n---- 开始执行工作流 ----")
start_time = time.time()# flow.run_sync() 会同步执行整个流程并返回最终步骤的结果# 对于更复杂的场景，可以使用 flow.run() (异步) 或 flow.stream() (流式处理)# 在Jupyter Notebook等已运行asyncio事件循环的环境中，应使用 await 调用异步方法try:
    final_result = await travel_flow.run() # context可以传递初始数据，这里我们简单处理
    
    print("---- 工作流执行完毕 ----")    print(f"总耗时: {time.time() - start_time:.2f} 秒")    
    print("\n========= 最终行程推荐 ==========")    if final_result:        print(final_result)        # 你也可以在这里用Ernie对结果进行润色或总结
        ernie_summary = get_ernie_response(f"请帮我总结以下旅行建议，使其更吸引人：\n{final_result}")        print("\n========= Ernie优化后的建议 ==========")        print(ernie_summary)    else:        print("未能获取到最终结果。")        print("\n工作流各步骤结果：")        for step_num, step_result in travel_flow.results.items():            print(f"步骤 {step_num}: {step_result}")except Exception as e:    print(f"工作流执行失败: {e}")    print("详细错误信息:")    import traceback
    traceback.print_exc()    print("\n工作流部分结果（如果存在）：")    if hasattr(travel_flow, 'results'):        for step_num, step_result in travel_flow.results.items():            print(f"步骤 {step_num}: {step_result}")

# 在Jupyter Notebook或主控脚本中# 假设 agent_network 中已注册了 ErnieKnowledgeAgent, ErnieAnalysisAgent, ErnieTranslationAgent# 它们都在各自的端口（如5006, 5007, 5008）上独立运行# parallel_flow = Flow(agent_network)# (parallel_flow.parallel() # 开始并行块#     .ask("ErnieKnowledgeAgent", "关于A2A协议的背景资料")#     .branch() # 新建一个并行分支#     .ask("ErnieAnalysisAgent", "分析这份销售数据：[数据...]，总结趋势。")#     .branch() # 再新建一个并行分支#     .ask("ErnieTranslationAgent", "请将这段英文翻译成中文：'The A2A protocol facilitates interoperability between AI agents.'")# .end_parallel()) # 结束并行块，等待所有分支完成# # 并行块的结果是一个字典，key是分支的序号(默认从'1'开始)或自定义名称，value是该分支最后一个操作的结果# # 例如 results_from_parallel['1'] 是知识问答结果, results_from_parallel['2'] 是数据分析结果等# # 接下来可以有一个串行步骤，汇总并行结果# def combine_results(results, context):#     knowledge_info = results['1'] # 上一步并行块整体是第1步#     analysis_report = knowledge_info["2"] # 获取并行块中第二个分支的结果#     translation = knowledge_info["3"]     # 第三个分支的结果#     # ... 结合这些信息生成最终报告#     # final_report = f"背景资料：{knowledge_info["1"]}\n数据分析：{analysis_report}\n翻译内容：{translation}"#     # return final_report#     # 注意：这里获取并行结果的方式需要根据实际 end_parallel() 返回的嵌套结构来调整#     # 通常 results['step_number_of_parallel_block'] 会是一个字典，包含各分支的结果#     # 例如: results_from_parallel = results['1'] # 假设并行块是第一步#     # background = results_from_parallel['1'] # 并行块的第一个分支#     # analysis = results_from_parallel['2'] # 并行块的第二个分支#     # translation_text = results_from_parallel['3'] # 并行块的第三个分支#     # return f"... {background} ... {analysis} ... {translation_text} ..."#     pass # 实际的组合逻辑# parallel_flow.execute_function(combine_results)# report = parallel_flow.run_sync()

# 假设你有 ernie_search_mcp_server.pypython ernie_search_mcp_server.py # 例如运行在 6001 端口# 假设你有 pdf_extractor_mcp_server.pypython pdf_extractor_mcp_server.py # 例如运行在 6002 端口

# 简化版 ernie_search_mcp_server.py 示例# from python_a2a.mcp import FastMCP, text_response, error_response# from ernie_client import get_ernie_response # 假设你想用Ernie来增强搜索# mcp_search_server = FastMCP(name="Ernie Search Tool", description="Performs web searches using Ernie")# @mcp_search_server.tool(#     name="web_search",#     description="Searches the web for a given query and returns top results."# )# def search_online(query: str, num_results: int = 3):#     """MCP tool function"""#     try:#         # 实际的搜索逻辑，例如调用搜索引擎API或 get_ernie_response

# SuperResearchAssistant 内部逻辑 - 简化示意# class SuperResearchAssistant(A2AServer):#     # ... 初始化时传入 mcp_client ...#     def handle_task(self, task: Task) -> Task:#         user_query = task.message.content.text#         keywords = self._extract_keywords(user_query) # 可能用Ernie#         # 1. MCP调用：搜索工具 (如 ErnieSearchTool on port 6001)#         try:#             raw_search_results = self.mcp_client.call("http://localhost:6001", "web_search", {"query": keywords})#             # search_items = parse_results(raw_search_results) #             search_items = [{"type":"pdf", "url":"url1", "title":"PDF1"}, {"type":"text", "snippet":"text1"}] # 模拟#         except Exception as e:#             return self._fail_task(task, f"搜索失败: {e}")#         # 2. MCP调用：内容提取 (如 PDFExtractor on port 6002 for PDFs)#         processed_content = []#         for item in search_items:#             if item["type"] == "pdf":#                 try:#                     # pdf_text = self.mcp_client.call("http://localhost:6002", "extract_pdf", {"url": item["url"]})#                     pdf_text = "模拟PDF文本 for " + item["title"]#                     processed_content.append(pdf_text)#                 except Exception:#                     processed_content.append(f"(无法提取PDF: {item['title']})")#             else:#                 processed_content.append(item["snippet"])#         full_text = "\n".join(processed_content)#         # 3. MCP调用：总结工具 (如 ErnieSummarizer on port 6003)#         try:#             # summary = self.mcp_client.call("http://localhost:6003", "summarize", {"text": full_text})#             # 为了演示，直接调用本地Ernie函数 (实际应通过MCP)#             from ernie_client import get_ernie_response#             summary = get_ernie_response(f"总结以下内容：{full_text[:1000]}", model="ernie-lite")#         except Exception as e:#             return self._fail_task(task, f"总结失败: {e}")#         task.artifacts = [{"parts": [{"type": "text", "text": summary}]}]#         task.status = TaskStatus(state=TaskState.COMPLETED)#         return task#     def _fail_task(self, task, error_message):#         task.status = TaskStatus(state=TaskState.FAILED, message=error_message)#         return task