A2A-over-MQTT Python SDK

Python SDK for the A2A-over-MQTT protocol: build agents and clients that communicate via MQTT using the A2A (Agent-to-Agent) protocol.

Install

pip install a2a-over-mqtt

Or with uv:

uv add a2a-over-mqtt

Quick start

Build a responder (agent)

import asyncio
from a2a_over_mqtt import MqttConfig, TopicSpace, Responder, build_card, A2ARequest

class MyAgent(Responder):
    async def on_request(self, request: A2ARequest, stream) -> str:
        await stream("thinking...")
        return f"You said: {request.text}"

async def main():
    mqtt = MqttConfig(host="localhost", port=1883)
    topics = TopicSpace(org="myorg", unit="default")
    card = build_card(
        name="Echo Agent",
        description="Echoes back whatever you send",
        url="mqtt://localhost:1883",
    )

    agent = MyAgent(
        agent_id="echo",
        mqtt=mqtt,
        topics=topics,
        card=card,
    )
    await agent.run()

asyncio.run(main())

Send a request (client)

import asyncio
from a2a_over_mqtt import MqttConfig, TopicSpace, Requester, A2ARequest

async def main():
    mqtt = MqttConfig(host="localhost", port=1883)
    topics = TopicSpace(org="myorg", unit="default")
    requester = Requester(mqtt, topics)

    req = A2ARequest(text="Hello, agent!", request_id="req-1")
    async for kind, content in requester.stream("echo", req.to_json(), "corr-1"):
        print(f"{kind}: {content}")

asyncio.run(main())

API overview

Configuration

  • MqttConfig(host, port, tls, username, password): MQTT broker connection parameters. Frozen dataclass.
  • TopicSpace(org, unit): A2A topic namespace. Provides .request(), .reply(), .discovery(), .event() methods.

Protocol (pure, no I/O)

  • A2ARequest / A2AResponse: JSON-RPC 2.0 message types with serialization.
  • make_status_event() / make_artifact_event(): Build streaming response messages.
  • classify_reply(data): Parse a reply into (kind, content) tuples.
  • validate_a2a_request(payload, response_topic, correlation): Validate inbound requests. Returns A2ARequest or ValidationError.

Discovery

  • build_card(name, description, url, ...): Build a spec-conformant A2A Agent Card.
  • parse_card(payload): Parse a card from JSON bytes or string.

High-level

  • Requester: Send requests and stream replies with configurable retry/timeout.
  • Responder: Abstract base class for agents. Implement on_request(), call run(). Handles the full protocol lifecycle: ack, streaming, artifact, terminal status, deduplication, cancellation, and LWT.

Development

uv sync
uv run python -m pytest tests/ -v

Build and publish

uv run python -m build
uv run twine upload dist/*

Project links

Data verified by PyPI on Apr 9, 2026
Data provided by the project maintainers, verified at the time the release was uploaded to PyPI.

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for hex6964 from gravatar.com hex6964

Credits

Author: Ivan Dyachkov

GitHub Statistics

Data verified by PyPI on Apr 9, 2026
The GitHub source repository was provided by the project maintainers and verified by PyPI at the time of upload. Stars, forks, and open issues/PRs are derived from that repository and have not been independently verified.

License expression

Apache-2.0
View SPDX License List

Requires

Python >=3.11

Classifiers

Report project as malware

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

a2a_over_mqtt-0.1.0.tar.gz (48.9 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

a2a_over_mqtt-0.1.0-py3-none-any.whl (19.7 kB view details)

Uploaded Python 3

File details

Details for the file a2a_over_mqtt-0.1.0.tar.gz.

File metadata

  • Download URL: a2a_over_mqtt-0.1.0.tar.gz
  • Upload date:
  • Size: 48.9 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for a2a_over_mqtt-0.1.0.tar.gz
Algorithm Hash digest
SHA256 9e75a8ae1e9930f03641f874d96e0f66845320ea878200143aa522ff6662d534
MD5 b196a4fb426846923269aba577cd11d6
BLAKE2b-256 c2eaf8c62bbf7b7db3de243c7d47f27a0a7d13744959c81d497ac866640f03de

See more details on using hashes here.

Provenance

The following attestation bundles were made for a2a_over_mqtt-0.1.0.tar.gz:

Publisher: sdk-python-publish.yml on emqx/mqtt-for-ai

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file a2a_over_mqtt-0.1.0-py3-none-any.whl.

File metadata

  • Download URL: a2a_over_mqtt-0.1.0-py3-none-any.whl
  • Upload date:
  • Size: 19.7 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/6.1.0 CPython/3.13.12

File hashes

Hashes for a2a_over_mqtt-0.1.0-py3-none-any.whl
Algorithm Hash digest
SHA256 b6f1dd95fe3ba9b46375793c27789cbc9e395ffe9a3b1023fc8c63fd5dcf2539
MD5 6bd5e25ceab2f45c49caa82fac89b057
BLAKE2b-256 9109ddaae71162e3cc22758d2c6a5e037149f60955508f453f3b4511bff88e52

See more details on using hashes here.

Provenance

The following attestation bundles were made for a2a_over_mqtt-0.1.0-py3-none-any.whl:

Publisher: sdk-python-publish.yml on emqx/mqtt-for-ai

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Release history Release notifications | RSS feed

This release

0.1.0 This release

2 files

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page