elkar-a2a

Elkar is an open-source task-management layer for AI agents — based on Google's Agent2Agent Protocol (A2A). Send, track, and orchestrate tasks across AI agents — effortlessly.

Elkar app     Website     💬 Discord    

✨ What is Elkar?

Elkar helps developers build collaborative, autonomous multi-agent systems with A2A protocol without the complexity of managing infrastructure.

Elkar provides:

• 🐍 A Python SDK (other languages coming soon) to build and connect AI agents using the A2A protocol with store and queue management.
• 🖥️ An application and API to manage, monitor and debug agent tasks.

Forget about infrastructure concerns—Elkar handles the complexity so your agents can focus on what matters: working together. Whether you're debugging agent behaviors or streaming tasks — Elkar makes A2A easy.

🔧 What can you do with Elkar?

Unlock seamless collaboration between your AI agents, whether they're in-house or external: Use it to:

• Effortlessly track and manage long-running tasks, with robust support for asynchronous operations via a persistent task store
• Browse and manage task history for observability and debugging
• Stream tasks between agents in real-time via dedicated SDKs
• Deeply debug agent tasks and A2A server interactions with full visibility on task history, artifacts, and server communications

Disclaimer: This project is still in early development.

👩‍💻 Software Development Kits

Elkar offers a Python SDK (with support for other languages coming soon) that simplifies the implementation of the A2A protocol.

Key features include:

• Complete implementation of the A2A protocol
• Task-oriented design focused on executing tasks, not managing infrastructure
• Built-in task management with simplified queuing and storage
• Support for streaming responses
• Custom authentication via RequestContext

🚀 Quickstart Onboarding

Follow the steps below to get started with Elkar Platform:

Step 1: Sign Up & Get Your API Key

Follow the steps below to get started with Elkar Platform:

1. Create an account on the Elkar platform.

2. Create a new agent: Navigate to Agents > Add a new agent.

3. Generate an API Key

   ⚠️ Copy the API key now — it won’t be shown again.

Step 2: Install Elkar SDK

pip install elkar

Step 3: 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 4: 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 5: Create your A2A Server

Instantiate your A2A Server with the Managed Service as Task Store.

api_key = "YOUR_ELKAR_API_KEY"  # Replace with your actual Elkar API key
store = ElkarClientStore(base_url="https://api.elkar.co/api", api_key=api_key)

task_manager: TaskManagerWithModifier = TaskManagerWithModifier(
    agent_card, 
    send_task_handler=task_handler,
    store=store
)

# 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

📦 Self-Hosting Package

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.

For detailed integration steps, see our Open Source Documentation

🛠️ A2A Debugger-Only Mode

If you want to run the client purely for debugging A2A servers without setting up authentication or task management:

1. Run in debugger-only mode

cd a2a-client
npm run a2adebugger

1. Open your browser at http://localhost:5173

This mode provides a simplified interface focused solely on debugging A2A-compatible servers without requiring Supabase authentication or backend server integration.

✅ Supported task updates

1. Status Update Describes the state of the task and the agent's progress. Messages in the status are appended to the task's history.

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

1. Artifact Update

Artifacts represent the result of the task. Indices are used to identify artifacts within a task. Updates append to existing artifacts if the index matches and the chunk is not the last one.

await task.upsert_artifact(
    Artifact(parts=[TextPart(text="I've finished the task!")], index=0)
)

1. Append Messages to History

Stores relevant information, such as thoughts or past communications, related to the task. (elkarbackup/elkarbackup-docker - GitHub)

await task.add_messages_to_history(
    [Message(parts=[TextPart(text="I'm working on the task...")])]
)

📚 Roadmap

• Full Documentation
• Task stores:
  • PostgreSQL, Redis, Hosted
• Task queues:
  • PostgreSQL, Redis, Hosted
• SDKs:
  • JavaScript/TypeScript
  • Go
  • Rust
• Tests and code samples
• Push notifications support
• Task history search functionality
• Integration with Model Context Protocol (MCP) for enhanced task management.

Check out our documentation for more details!

💬 Community

Join our Discord server to get help, share ideas, and get updates

🤝 Contribute

We ❤️ feedback, PRs, and ideas! Here's how to help:

• If you find Elkar useful, a GitHub ⭐️ would mean a lot! — it helps to support the project!
• Report bugs or request features via issues.
• Show off what you've built with Elkar here!
• Submit pull requests, and we'll review it as soon as possible.

🔒 License

This project is licensed under the MIT License – see the LICENSE file for details.