A2A Gateway Adapter

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

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