by elkar-ai
Elkar - Task management for AI Agents, ready for Agent2Agent (A2A) protocol
# Add to your Claude Code skills
git clone https://github.com/elkar-ai/elkar-a2aNo comments yet. Be the first to share your thoughts!
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 helps developers build collaborative, autonomous multi-agent systems with A2A protocol without the complexity of managing infrastructure.
Elkar provides:
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.
Unlock seamless collaboration between your AI agents, whether they're in-house or external: Use it to:
Disclaimer: This project is still in early development.
Elkar offers a Python SDK (with support for other languages coming soon) that simplifies the implementation of the A2A protocol.
Key features include:
RequestContextFollow the steps below to get started with Elkar Platform:
Follow the steps below to get started with Elkar Platform:
Create an account on the Elkar platform.
Create a new agent: Navigate to Agents > Add a new agent.
Generate an API Key
⚠️ Copy the API key now — it won’t be shown again.
pip install elkar
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,
)
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,
),
)
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
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
If you want to run the client purely for debugging A2A servers without setting up authentication or task management:
cd a2a-client
npm run a2adebugger
http://localhost:5173This mode provides a simplified interface focused solely on debugging A2A-compatible servers without requiring Supabase authentication or backend server integration.
await task.set_status(
TaskStatus(
state=TaskState.COMPLETED,
message=Message(parts=[TextPart(text="I've finished the task!")])
)
)
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)
)
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...")])]
)
Check out our documentation for more details!
Join our Discord server to get help, share ideas, and get updates
We ❤️ feedback, PRs, and ideas! Here's how to help:
This project is licensed under the MIT License – see the LICENSE file for details.