打造AI協作未來：淺談MCP、A2A與ACP的角色與應用 與ACP實戰

%%writefile ../my_acp_project/crew_agent_server.py #輸出文件...#要餵進去RAG的文檔rag_tool.add("../data/gold-hospital-and-premium-extras.pdf", data_type="pdf_file")@server.agent()async def policy_agent(input: list[Message]) -> AsyncGenerator[RunYield, RunYieldResume]:    "This is an agent for questions around policy coverage, it uses a RAG pattern to find answers based on policy documentation. Use it to help answer questions on coverage and waiting periods."    insurance_agent = Agent(        role="Senior Insurance Coverage Assistant",         goal="Determine whether something is covered or not",        backstory="You are an expert insurance agent designed to assist with coverage queries",        verbose=True,        allow_delegation=False,        llm=llm,        tools=[rag_tool],         max_retry_limit=5    )        task1 = Task(         description=input[0].parts[0].content,         expected_output = "A comprehensive response as to the users question",         agent=insurance_agent    )    crew = Crew(agents=[insurance_agent], tasks=[task1], verbose=True)        task_output = await crew.kickoff_async()    yield Message(parts=[MessagePart(content=str(task_output))])if __name__ == "__main__":    server.run(port=8001)

%%writefile ../my_acp_project/smolagents_server.py...#建立另一隻不同架構的CodeAgent Agent @server.agent()async def health_agent(input: list[Message], context: Context) -> AsyncGenerator[RunYield, RunYieldResume]:    "This is a CodeAgent which supports the hospital to handle health based questions for patients. Current or prospective patients can use it to find answers about their health and hospital treatments."    agent = CodeAgent(tools=[DuckDuckGoSearchTool(), VisitWebpageTool()], model=model)    prompt = input[0].parts[0].content    response = agent.run(prompt)    yield Message(parts=[MessagePart(content=str(response))])if __name__ == "__main__":    server.run(port=8000)

#當兩隻ACP Agent都架設好後from acp_sdk.client import Clientimport asynciofrom colorama import Fore #可以手動的去呼叫兩隻ACP agent，適用於需要遵循特定流程的應用async def run_hospital_workflow() -> None:    async with Client(base_url="http://localhost:8001") as insurer, Client(base_url="http://localhost:8000") as hospital:        run1 = await hospital.run_sync(            agent="health_agent", input="Do I need rehabilitation after a shoulder reconstruction?"        )        content = run1.output[0].parts[0].content        print(Fore.LIGHTMAGENTA_EX+ content + Fore.RESET)        run2 = await insurer.run_sync(            agent="policy_agent", input=f"Context: {content} What is the waiting period for rehabilitation?"        )        print(Fore.YELLOW + run2.output[0].parts[0].content + Fore.RESET)asyncio.run(run_hospital_workflow())

#或是如下，把agents 做成一個Jason放進acp_agents，再透過acpagent去決定拿到的task應該呼叫哪個Agentasync def run_hospital_workflow() -> None:    async with Client(base_url="http://localhost:8001") as insurer, Client(base_url="http://localhost:8000") as hospital:        # agents discovery        agent_collection = await AgentCollection.from_acp(insurer, hospital)          acp_agents = {agent.name: {'agent':agent, 'client':client} for client, agent in agent_collection.agents}        print(acp_agents)         # passing the agents as tools to ACPCallingAgent        acpagent = ACPCallingAgent(acp_agents=acp_agents, model=model)        # running the agent with a user query        result = await acpagent.run("do i need rehabilitation after a shoulder reconstruction and what is the waiting period from my insurance?")        print(Fore.YELLOW + f"Final result: {result}" + Fore.RESET)