a2a-json-rpc

A JSON-RPC implementation for the A2A (Agent-to-Agent) Protocol.

Overview

The a2a-json-rpc library provides a lightweight, transport-agnostic implementation of the JSON-RPC 2.0 protocol specifically tailored for A2A (Agent-to-Agent) communication. It includes:

• Complete JSON-RPC 2.0 request/response handling
• A2A-specific error definitions and handling
• Pydantic models for type-safe message processing
• Async/await support via anyio
• Transport abstraction layer

Installation

pip install a2a-json-rpc

Quick Start

import asyncio
from a2a_json_rpc.protocol import JSONRPCProtocol
from a2a_json_rpc.models import Json

# Create a protocol instance
protocol = JSONRPCProtocol()

# Register a method handler
@protocol.method("echo")
async def echo_handler(method: str, params: Json) -> Json:
    return params

# Process a request
async def main():
    request = protocol.create_request("echo", {"message": "Hello, A2A!"})
    response = await protocol._handle_raw_async(request)
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

Features

• Type Safety: Built with Pydantic for robust data validation and serialization
• Async First: Designed for asynchronous communication patterns
• Transport Agnostic: Can be used with HTTP, WebSockets, or any other transport
• Error Handling: Comprehensive error types for both JSON-RPC and A2A-specific errors
• Notifications: Support for fire-and-forget notifications

A2A Protocol Support

The library implements the A2A Protocol specification, which defines a standard way for AI agents to communicate with each other. The implementation includes:

• Task management operations
• Push notification handling
• Streaming support
• Message history tracking

Error Handling

The library provides a comprehensive set of error types:

from a2a_json_rpc.json_rpc_errors import ParseError, InvalidRequestError
from a2a_json_rpc.a2a_errors import TaskNotFoundError, PushNotificationsNotSupportedError

# JSON-RPC standard errors
raise ParseError("Invalid JSON payload")
raise InvalidRequestError("Missing required field")

# A2A-specific errors
raise TaskNotFoundError(data={"id": "task-123"})
raise PushNotificationsNotSupportedError()

Transport Layer

The transport layer is defined as a protocol interface, allowing for different implementations:

from a2a_json_rpc.transport import JSONRPCTransport

class MyTransport(JSONRPCTransport):
    async def call(self, method: str, params: any) -> any:
        # Implementation
        ...
    
    async def notify(self, method: str, params: any) -> None:
        # Implementation
        ...
    
    def stream(self) -> AsyncIterator[Json]:
        # Implementation
        ...

Development

Setup

# Clone the repository
git clone https://github.com/yourusername/a2a-json-rpc.git
cd a2a-json-rpc

# Install development dependencies
make dev-install

Testing

make test

Building

make build

Publishing

make publish

License

MIT

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add some amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request