Wrap Your Existing Agents

uv add agentstack-sdk

# Import your existing agent logic
from my_agent import run_my_agent  # Your existing agent code

import os
from a2a.types import Message
from a2a.utils.message import get_message_text
from agentstack_sdk.server import Server
from agentstack_sdk.server.context import RunContext
from agentstack_sdk.a2a.types import AgentMessage

server = Server()

@server.agent()
async def my_wrapped_agent(input: Message, context: RunContext):
    """Wrapper around my existing agent"""
    
    # Extract the user's message
    user_message = get_message_text(input)
    
    # Call your existing agent logic
    # This can be synchronous or asynchronous
    result = await run_my_agent(user_message)
    
    # Yield the response back to Agent Stack
    yield AgentMessage(text=result)

def run():
    server.run(
        host=os.getenv("HOST", "127.0.0.1"),
        port=int(os.getenv("PORT", 8000))
    )

if __name__ == "__main__":
    run()

uv run server.py

@server.agent()
async def streaming_agent(input: Message, context: RunContext):
    user_message = get_message_text(input)
    
    # Stream results as they come
    async for chunk in my_streaming_agent(user_message):
        yield AgentMessage(text=chunk)

@server.agent()
async def contextual_agent(input: Message, context: RunContext):
    # Get conversation history
    previous_messages = context.history
    
    # Your agent can use this context
    result = await my_agent_with_context(
        current_message=get_message_text(input),
        history=previous_messages
    )
    
    yield AgentMessage(text=result)