Metadata-Version: 2.4
Name: a2a-min
Version: 0.1.0
Summary: Agent to Agent (A2A) minimalistic Python SDK
Author: A2A SDK Team
Project-URL: Homepage, https://github.com/pcingola/a2a_min
Project-URL: Bug Tracker, https://github.com/pcingola/a2a_min/issues
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12.2
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: dotenv>=0.9.9
Requires-Dist: httpx>=0.28.1
Requires-Dist: httpx-sse>=0.4.0
Requires-Dist: ipykernel>=6.29.5
Requires-Dist: jwcrypto>=1.5.6
Requires-Dist: jwt>=1.3.1
Requires-Dist: pydantic>=2.11.3
Requires-Dist: pyjwt>=2.10.1
Requires-Dist: ruff>=0.11.5
Requires-Dist: sse-starlette>=2.2.1
Requires-Dist: starlette>=0.46.2
Requires-Dist: uvicorn>=0.34.1
Dynamic: license-file

# A2A Minimalistic Python SDK

A minimalistic Python SDK for [Agent-to-Agent (A2A)](https://google.github.io/A2A/#/) communication.

## Installation

You can install the package from PyPI using pip:

```bash
# Using pip
pip install a2a-min

# Using uv
uv pip install a2a-min
```
## Overview

The A2A Min SDK provides a Python implementation of the Agent-to-Agent (A2A) protocol, enabling communication between AI agents. The package includes:

- A low-level API for direct interaction with the A2A protocol
- A high-level abstraction layer for simplified usage
- Support for streaming responses
- Support for multimodal content (text, images, structured data)
- Middleware for cross-cutting concerns like logging and metrics

## Installing From GitHub

To install the latest version directly from GitHub:

```bash
# Using pip
pip install git+https://github.com/pcingola/a2a_min.git

# Using uv
uv pip install git+https://github.com/pcingola/a2a_min.git
```

## Basic Usage

### Creating a Simple Echo Agent Server

Here's how to create a simple server that hosts an echo agent:

```python
from a2a_min import AgentAdapter, A2aMinServer, AgentInvocationResult

class EchoAgent(AgentAdapter):
    """ A simple echo agent that repeats the user's message """
    def invoke(self, query: str, session_id: str) -> AgentInvocationResult:
        """Echo back the user's query."""
        return AgentInvocationResult.agent_msg(f"Echo: {query}")

# Start the echo agent server
A2aMinServer.from_agent(EchoAgent()).start()
```

Save this as `echo_a2a_server.py` and run it to start the server:

```bash
python a2a_min/examples/echo_a2a/echo_a2a_server.py
```

### Creating a Client to Interact with the Agent

Here's how to create a client that connects to the echo agent:

```python
import asyncio
from a2a_min import A2aMinClient

async def client():
    """ Run the example client """
    client = A2aMinClient.connect("http://localhost:8000")
    task = await client.send_message("Hello, Echo Agent!")    
    # Print the response
    for artifact in task.artifacts:
        for part in artifact.parts:
            if hasattr(part, "text"):
                print(f"Response: {part.text}")

if __name__ == "__main__":
    # Run the client
    asyncio.run(client())
```

Save this as `echo_a2a_client.py` and run it to interact with the server:

```bash
 python a2a_min/examples/echo_a2a/echo_a2a_client.py
```

## Creating Custom Agents

To create your own agent, extend the `AgentAdapter` class and implement the `invoke` method:

```python
from a2a_min import AgentAdapter, AgentInvocationResult

class MyCustomAgent(AgentAdapter):
    def invoke(self, query: str, session_id: str) -> AgentInvocationResult:
        # Process the query and generate a response
        response = f"You asked: {query}. Here's my response..."
        return AgentInvocationResult.agent_msg(response)
```

## Advanced Features

### Streaming Responses

To create an agent that streams its response:

```python
import asyncio
from a2a_min import AgentAdapter, AgentInvocationResult

class StreamingAgent(AgentAdapter):
    async def stream(self, query: str, session_id: str):
        words = f"You asked: {query}. This is a streaming response.".split()
        
        for i, word in enumerate(words):
            partial_text = " ".join(words[:i+1])
            
            yield AgentInvocationResult.agent_msg(
                partial_text,
                is_complete=(i == len(words) - 1)
            )
            
            await asyncio.sleep(0.2)
```

To receive streaming responses in the client:

```python
async def streaming_client():
    client = A2aMinClient.connect("http://localhost:8000")
    async for update in client.send_message_streaming("Hello, Streaming Agent!"):
        if update.artifact:
            for part in update.artifact.parts:
                if hasattr(part, "text"):
                    print(f"Partial response: {part.text}")
```

### Multimodal Support

To create an agent that can respond with text, images, and structured data:

```python
from a2a_min import AgentAdapter, AgentInvocationResult
from a2a_min.types import Message, TextPart, FilePart, FileContent, DataPart

class MultiModalAgent(AgentAdapter):
    def invoke(self, query: str, session_id: str) -> AgentInvocationResult:
        # Create a multimodal response with text, image, and data
        message = Message(
            role="agent",
            parts=[
                TextPart(text="Here's an image and some data:"),
                FilePart(
                    file=FileContent(
                        name="example.jpg",
                        mimeType="image/jpeg",
                        uri="https://example.com/image.jpg"
                    )
                ),
                DataPart(
                    data={
                        "temperature": 72.5,
                        "unit": "Fahrenheit",
                        "conditions": "Sunny"
                    }
                )
            ]
        )
        
        return AgentInvocationResult(
            message=message,
            is_complete=True,
            requires_input=False
        )
```

### Using Middleware

You can add middleware to your server for logging, metrics, or other cross-cutting concerns:

```python
from a2a_min import A2aMinServer, LoggingMiddleware, MetricsMiddleware

# Create middleware
logging_middleware = LoggingMiddleware()

def record_metric(name, value):
    print(f"Metric {name}: {value}")

metrics_middleware = MetricsMiddleware(record_metric)

# Create server with middleware
server = A2aMinServer.from_agent(
    MyCustomAgent(),
    middlewares=[logging_middleware, metrics_middleware]
)
server.start()
```

## Examples

The package includes several examples to demonstrate its usage:

- `a2a_min/examples/echo_a2a`: A simple echo agent example
- `a2a_min/examples/streaming_agent_example.py`: An agent that streams its response
- `a2a_min/examples/multimodal_agent_example.py`: An agent that responds with text, images, and data
- `a2a_min/examples/base`: Low-level API examples

To run the echo example:

1. Start the server in one terminal:
   ```bash
   python -m a2a_min.examples.echo_a2a.echo_a2a_server
   ```

2. Run the client in another terminal:
   ```bash
   python -m a2a_min.examples.echo_a2a.echo_a2a_client
   ```

## Documentation

For more detailed documentation, see:

- `a2a_min/docs/abstraction_readme.md`: Overview of the abstraction layer
- `a2a_min/docs/new_abstraction.md`: Detailed design of the abstraction layer

## Benefits of Using a2a_min

- **Simplified API**: The abstraction layer provides a much simpler API for both client and server sides
- **Type Safety**: All components use Pydantic models for strong typing
- **Extensibility**: Clear extension points for middleware, logging, etc.
- **Optional Features**: Streaming and push notifications are optional but easy to enable
- **Consistent Naming**: Method names are aligned with current type names
- **Reuse Existing Types**: Leverages the current type system for compatibility
