i2y/castella/castella-a2a

from castella.a2a import A2AClient
client = A2AClient("http://agent.example.com")
print(f"Connected to: {client.name}")
print(f"Skills: {[s.name for s in client.skills]}")
response = client.ask("What's the weather in Tokyo?")
print(response)

uv sync --extra agent   # A2A + A2UI support

from castella.a2a import A2AClient
client = A2AClient(
    agent_url="http://localhost:8080",
    timeout=30.0,
)

# Synchronous ask
response = client.ask("Hello!")
# Asynchronous ask
response = await client.ask_async("Hello!")
# Streaming (async)
async for chunk in client.ask_stream("Tell me a story"):
    print(chunk, end="", flush=True)
# Send structured message
from castella.a2a import Message
response = client.send_message(Message(role="user", content="Hello"))
# Check skills
if client.has_skill("get_weather"):
    skill = client.get_skill("get_weather")
    print(f"Skill: {skill.name} - {skill.description}")

from castella.a2a import A2AClient
client = A2AClient("http://agent.example.com")
card = client.agent_card
print(f"Name: {card.name}")
print(f"Description: {card.description}")
print(f"Version: {card.version}")
print(f"URL: {card.url}")
# List skills
for skill in card.skills:
    print(f"  - {skill.name}: {skill.description}")
    print(f"    Tags: {skill.tags}")

skill = client.get_skill("get_weather")
print(skill.name)          # "get_weather"
print(skill.description)   # "Get current weather"
print(skill.tags)          # ["weather", "api"]
print(skill.examples)      # ["What's the weather in Tokyo?"]

from castella.a2a import A2AClient, A2AConnectionError, A2AResponseError
try:
    client = A2AClient("http://agent.example.com")
    response = client.ask("Hello")
except A2AConnectionError as e:
    print(f"Connection failed: {e}")
except A2AResponseError as e:
    print(f"Agent error: {e}")

async def stream_story():
    client = A2AClient("http://agent.example.com")
    if client.supports_streaming:
        async for chunk in client.ask_stream("Tell me a story"):
            print(chunk, end="", flush=True)
    else:
        # Fallback to non-streaming
        response = await client.ask_async("Tell me a story")
        print(response)

from castella import App, Column
from castella.agent import AgentCardView
from castella.a2a import A2AClient
from castella.frame import Frame
client = A2AClient("http://agent.example.com")
card_view = AgentCardView(
    client.agent_card,
    show_skills=True,
    compact=False,
)
App(Frame("Agent Info", 400, 300), card_view).run()

from castella.agent import AgentChat
# Automatic connection and chat UI
chat = AgentChat.from_a2a("http://localhost:8080")
chat.run()

from python_a2a import A2AServer, skill, run_server
class WeatherAgent(A2AServer):
    @skill(name="get_weather", description="Get current weather")
    def get_weather(self, location: str) -> str:
        return f"Weather in {location}: Sunny, 22°C"
agent = WeatherAgent(
    name="Weather Agent",
    description="Provides weather information",
    version="1.0.0",
    url="http://localhost:8080",
)
run_server(agent, port=8080)