agent-client-protocol

import asyncio from acp.agents import Agent from acp.schemas import Message, MessageContents, TextResourceContents, MessageID class SimpleEchoAgent(Agent): """ A simple agent that echoes back any text message it receives. In a real application, this agent would be registered with an AgentServer and communicate via a transport layer (e.g., stdio). """ # In a full setup, send_message would use the transport for communication. # We override it here purely for demonstration purposes in this quickstart. async def send_message(self, message: Message) -> None: if message.contents and message.contents.text: print(f"Agent sending: '{message.contents.text.text}'") async def handle_message(self, message: Message) -> None: if message.contents and message.contents.text: response_text = f"Echo from agent: {message.contents.text.text}" response_message = Message( id=MessageID.new_id(), contents=MessageContents( text=TextResourceContents(text=response_text) ), parent_id=message.id, ) await self.send_message(response_message) else: print(f"Agent received non-text message: {message.id}") async def run_agent_example(): agent = SimpleEchoAgent() # Simulate an incoming message that a real server would pass to the agent incoming_message = Message( id=MessageID.new_id(), contents=MessageContents( text=TextResourceContents(text="Hello ACP!") ) ) print(f"Simulating client sending: '{incoming_message.contents.text.text}'\n") await agent.handle_message(incoming_message) print("\nAgent quickstart example finished.") if __name__ == "__main__": asyncio.run(run_agent_example())