A2A Gateway Adapter

1
import asyncio
2
import os
3
4
from dotenv import load_dotenv
5
from band import Agent
6
from band.adapters import A2AGatewayAdapter
7
from band.config import load_agent_config
8
9
async def main():
10
    load_dotenv()
11
    agent_id, api_key = load_agent_config("my_agent")
12
13
    adapter = A2AGatewayAdapter(
14
        gateway_url="http://localhost:10000",
15
        port=10000,
16
    )
17
18
    agent = Agent.create(
19
        adapter=adapter,
20
        agent_id=agent_id,
21
        api_key=api_key,
22
        ws_url=os.getenv("BAND_WS_URL", "wss://app.band.ai/api/v1/socket/websocket"),
23
        rest_url=os.getenv("BAND_REST_URL", "https://app.band.ai"),
24
    )
25
26
    print("Gateway running on http://localhost:10000")
27
    await agent.run()
28
29
if __name__ == "__main__":
30
    asyncio.run(main())

1
from a2a.client import ClientConfig, create_client
2
from a2a.helpers import get_message_text
3
from a2a.types import Message, Part, Role, SendMessageRequest, TaskState
4
5
async def main():
6
    client = await create_client(
7
        agent="http://localhost:10000/agents/weather-agent",
8
        client_config=ClientConfig(streaming=True),
9
    )
10
11
    request = SendMessageRequest(
12
        message=Message(
13
            message_id="msg-001",
14
            context_id="conversation-123",
15
            role=Role.ROLE_USER,
16
            parts=[Part(text="What is the weather in New York?")],
17
        )
18
    )
19
20
    async for response in client.send_message(request):
21
        if response.task.status.state == TaskState.TASK_STATE_COMPLETED:
22
            print(f"Response: {get_message_text(response.task.status.message)}")