You can use Elkar self-hosted version as a simple library with implemented task management and streaming in local memory. Support for other task stores as PostgreSQL or Redis will come soon.

Checkout our GitHub repository

Follow the steps below to get started with Elkar Open Source:

​
Step 1: Install Elkar SDK

pip install elkar

​
Step 2: Create your Task Handler

The task handler manages a task’s status and artifacts, abstracting away the complexity of interacting with the Task Store. It supports three operations: set_status, add_messages_to_history, and upsert_artifact.

The task handler currently has a strict signature: async def my_handler(task: TaskModifierBase, request_context: RequestContext) -> None.While this signature is enforced for now, it may become more flexible in the future.

from elkar.a2a_types import *
from elkar.server.server import A2AServer
from elkar.task_manager.task_manager_base import RequestContext
from elkar.task_manager.task_manager_with_task_modifier import TaskManagerWithModifier
from elkar.task_modifier.base import TaskModifierBase

async def task_handler(
    task: TaskModifierBase, request_context: RequestContext | None
) -> None:

    await task.set_status(
        TaskStatus(
            state=TaskState.WORKING,
            message=Message(
                role="agent",
                parts=[TextPart(text="I understand the task, I'm working on it...")],
            ),
        )
    )

    await task.upsert_artifacts(
        [
            Artifact(
                parts=[TextPart(text="I've finished the task, here is the result...")],
                index=0,
            )
        ]
    )

    await task.set_status(
        TaskStatus(
            state=TaskState.COMPLETED,
            message=Message(
                role="agent",
                parts=[TextPart(text="I've finished the task!")],
            ),
        ),
        is_final=True,
    )

​
Step 3: Create your Agent Card

Your Agent Card is defined following the requirements from A2A Protocol.

agent_card = AgentCard(
    name="Test Agent",
    description="Test Agent Description",
    url="https://example.com",
    provider=AgentProvider(organization="Elkar", url="https://www.elkar.co"),
    documentationUrl="https://example.com/documentation",
    version="1.0.0",
    skills=[
        AgentSkill(
            id="1",
            name="Generate image",
            description="Generate images from prompt description",
            inputModes=["text"],
            outputModes=["image"],
        ),
    ],
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=True,
        stateTransitionHistory=True,
    ),
)

​
Step 4: Create your A2A Server

Instantiate your A2A Server with In Memory Task Store.

task_manager: TaskManagerWithModifier = TaskManagerWithModifier(
    agent_card, 
    send_task_handler=task_handler
    # Optionally, configure a store here (e.g., for managed service or custom persistence)
)

# Create the server instance
server = A2AServer(task_manager, host="0.0.0.0", port=5001, endpoint="/")

server.start() # This is blocking. For production, use an ASGI server like Uvicorn.

To run this example (e.g., if saved as main.py and you expose server.app as app): uvicorn main:app —host 0.0.0.0 —port 5001

uvicorn main:app --host 0.0.0.0 --port 5001