python-base-agent 2026.2.13
Python implementation of BaseAgent for distributed message processing
Requires Python: >=3.12

# Python BaseAgent

Python implementation of the BaseAgent framework for building distributed message processing agents with exactly-once semantics.

## Overview

BaseAgent provides a robust foundation for building agents that:
- Process messages from Kafka with exactly-once semantics
- Manage state with pluggable storage backends (Redis/Memory)
- Store large payloads in object storage (S3/MinIO)
- Provide health checks for Kubernetes deployments
- Include distributed tracing with OpenTelemetry
- Handle deduplication with vector clocks

## Installation

The platform uses `mise` for task management and `uv` for Python dependency management:

```bash
# Using mise tasks (recommended - platform convention)
mise run base-agent-install

# Or manually with uv
uv pip install -e ".[dev]"

# Or with pip
pip install -e ".[dev]"
```

## Quick Start

```python
from base_agent import BaseAgent, AgentConfig, Envelope

class WordCountAgent(BaseAgent):
    """Example agent that counts words in messages"""

    async def process_message(self, envelope: Envelope):
        """Process a single message"""
        text = envelope.payload.get("text", "")
        word_count = len(text.split())

        # Store result in object storage
        await self.store_result(f"count-{envelope.header.message_id}", {
            "message_id": envelope.header.message_id,
            "word_count": word_count,
            "timestamp": envelope.header.send_timestamp
        })

        # Send notification
        await self.send_notification(
            envelope.header,
            f"Processed {word_count} words"
        )

# Configure and run
config = AgentConfig(
    kafka_brokers="localhost:9092",
    consumer_group="word-count-group",
    topics=["input-topic"],
    object_store_endpoint="http://localhost:9000",
    object_store_access_key="admin",
    object_store_secret_key="password",
    redis_url="redis://localhost:6379"
)

async def main():
    async with WordCountAgent("word-count", config) as agent:
        await agent.run()  # Runs until interrupted
```

## Architecture

BaseAgent follows the same architecture as the Java BaseAgent:

```
┌─────────────────────────────────────────────────┐
│                  BaseAgent                       │
├─────────────────────────────────────────────────┤
│  Kafka Consumer │ Message Processing │ Producer │
├─────────────────────────────────────────────────┤
│  State Store    │ Object Store    │ Health API  │
├─────────────────────────────────────────────────┤
│  Vector Clock   │ Telemetry      │ Metrics      │
└─────────────────────────────────────────────────┘
```

## Core Components

### 1. Message Processing
- Kafka consumer with manual offset management
- Transactional producer for exactly-once semantics
- Automatic deduplication using vector clocks
- Session tracking to prevent reprocessing

### 2. State Management
- Pluggable state stores (Redis, In-Memory)
- Offset tracking for fault tolerance
- Session high-watermark tracking
- Atomic operations for consistency

### 3. Object Storage
- S3-compatible object storage (S3, MinIO) and Azure Blob Storage
- Large payload handling with streaming support
- Automatic bucket creation and validation
- Full CRUD operations (put, get, delete, list, copy)
- Presigned URL generation for temporary access
- Object metadata management
- Pagination support for listing large buckets
- Path-style access for MinIO compatibility

### 4. Health Monitoring
- Kubernetes-compatible health endpoints
- Liveness: `/health/liveness`
- Readiness: `/health/readiness`
- Configurable port (set to 0 to disable)

### 5. Observability
- OpenTelemetry tracing
- Prometheus metrics
- Structured logging with context
- Distributed trace propagation

## Configuration

Configuration can be provided via:
1. Environment variables
2. `.env` file
3. Direct configuration object

### Required Configuration

```python
config = AgentConfig(
    # Kafka (required)
    kafka_brokers="localhost:9092",
    consumer_group="my-group",
    topics=["input-topic"],

    # Object Storage (required) - see provider-specific config below
    object_store_bucket="agent-data",
)
```

### Object Store Configuration

The BaseAgent supports multiple object store providers: S3, MinIO, and Azure Blob Storage.

#### Common Configuration

```bash
# Provider type: s3, minio, or azure (default: s3)
OBJECT_STORE_TYPE=s3

# Bucket/Container name (used by all providers)
OBJECT_STORE_BUCKET=agent-data
```

#### S3/MinIO Configuration

```bash
OBJECT_STORE_TYPE=s3  # or minio
OBJECT_STORE_BUCKET=agent-data

# Object Store Endpoint (for S3-compatible stores like MinIO)
OBJECT_STORE_ENDPOINT=http://localhost:9000

# Object Store Credentials
OBJECT_STORE_ACCESS_KEY=admin
OBJECT_STORE_SECRET_KEY=password
OBJECT_STORE_REGION=us-east-1
```

**Development (MinIO):**
```bash
OBJECT_STORE_TYPE=minio
OBJECT_STORE_ENDPOINT=http://localhost:9000
OBJECT_STORE_ACCESS_KEY=admin
OBJECT_STORE_SECRET_KEY=password
OBJECT_STORE_BUCKET=agent-data
```

**Production (AWS S3):**
```bash
OBJECT_STORE_TYPE=s3
# Leave OBJECT_STORE_ENDPOINT unset to use AWS S3 default
OBJECT_STORE_ACCESS_KEY=your-aws-access-key
OBJECT_STORE_SECRET_KEY=your-aws-secret-key
OBJECT_STORE_BUCKET=your-production-bucket
```

#### Azure Blob Storage Configuration

Azure Blob Storage can be configured using one of three authentication methods:

**Option 1: Azure AD (DefaultAzureCredential) - Recommended for production:**
```bash
OBJECT_STORE_TYPE=azure
OBJECT_STORE_BUCKET=agent-data
AZURE_STORAGE_ACCOUNT=yourstorageaccount
# Optional: AZURE_STORAGE_ENDPOINT=https://youracct.blob.core.windows.net
```

This uses Azure's DefaultAzureCredential chain (environment variables, managed identity, workload identity, etc.).

**Option 2: Connection String:**
```bash
OBJECT_STORE_TYPE=azure
OBJECT_STORE_BUCKET=agent-data
AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...
```

**Option 3: Account Key:**
```bash
OBJECT_STORE_TYPE=azure
OBJECT_STORE_BUCKET=agent-data
AZURE_STORAGE_ACCOUNT=yourstorageaccount
AZURE_STORAGE_KEY=your-storage-account-key
```

**Development (Azurite):**
```bash
OBJECT_STORE_TYPE=azure
OBJECT_STORE_BUCKET=agent-data
AZURE_STORAGE_CONNECTION_STRING=UseDevelopmentStorage=true;
```

#### Presigned URLs (SAS) Behavior

For Azure Blob Storage, presigned URLs are implemented using Shared Access Signatures (SAS):

- **Account Key auth**: SAS tokens are signed directly with the account key
- **Azure AD auth**: User Delegation SAS tokens are generated using a delegation key obtained from Azure AD
  - Requires the service principal to have the `Microsoft.Storage/storageAccounts/blobServices/generateUserDelegationKey/action` permission (included in the Storage Blob Data Contributor role)
- **Connection String auth**: SAS tokens are signed with the account key extracted from the connection string

If SAS generation is not possible (e.g., missing permissions), `generate_presigned_url` returns `None`.

### Optional Configuration

```python
config = AgentConfig(
    # ... required fields ...

    # State Storage (defaults to in-memory)
    state_store_type="redis",  # or "memory"
    redis_url="redis://localhost:6379",

    # Health Check
    health_check_port=8080,  # Set to 0 to disable

    # Admin Gateway Registration
    admin_gateway_url="http://localhost:8888",

    # Telemetry
    otel_endpoint="http://localhost:4317",

    # Performance
    kafka_max_poll_records=100,
    kafka_poll_timeout_ms=100,
)
```

## Testing

The project follows Test-Driven Development (TDD) principles.

### Running Tests with Mise (Platform Convention)

```bash
# Using mise tasks (recommended)
mise run base-agent-test        # Run all tests
mise run base-agent-test-unit   # Run unit tests only
mise run base-agent-test-int    # Run integration tests

# Format and lint
mise run base-agent-fmt         # Format code
mise run base-agent-lint        # Lint code
```

### Running Tests Manually

```bash
# Run all tests
pytest

# Run unit tests only
pytest tests/unit -m unit

# Run integration tests (requires Docker)
pytest tests/integration -m integration

# Run with coverage
pytest --cov=base_agent --cov-report=html

# Run specific test file
pytest tests/unit/test_agent.py

# Run tests in parallel
pytest -n auto
```

### Test Structure

```
tests/
├── unit/           # Fast, isolated unit tests
│   ├── test_agent.py
│   ├── test_state_store.py
│   ├── test_kafka_client.py
│   ├── test_object_store.py    # ✅ IMPLEMENTED
│   └── test_vector_clock.py
├── integration/    # Tests with real services
│   ├── test_end_to_end.py
│   ├── test_exactly_once.py
│   └── test_recovery.py
└── conftest.py     # Shared fixtures
```

### Integration Testing with TestContainers

Integration tests use testcontainers to spin up real services:

```python
# tests/conftest.py
@pytest.fixture(scope="session")
async def kafka():
    with KafkaContainer() as kafka:
        yield kafka.get_bootstrap_server()

@pytest.fixture(scope="session")
async def redis():
    with RedisContainer() as redis:
        yield redis.get_connection_url()
```

## Development

### Setup Development Environment

```bash
# Clone the repository
git clone <repo>
cd python-base-agent

# Install dependencies using mise (platform standard)
mise run base-agent-install

# Start development services
mise run base-agent-docker-up
```

### Code Quality

All code quality tasks are managed through mise (no Makefile/custom scripts):

```bash
# Format code
mise run base-agent-fmt

# Lint code
mise run base-agent-lint

# Type checking
mise run base-agent-type-check

# All quality checks
mise run base-agent-quality
```

### Building Documentation

```bash
# Install docs dependencies
pip install -e ".[docs]"

# Build HTML docs
cd docs
make html

# View docs
open _build/html/index.html
```

## Performance

The Python BaseAgent is designed to match the performance characteristics of the Java version:

- **Throughput**: 10,000+ messages/second per agent
- **Latency**: < 10ms p99 message processing
- **Memory**: < 200MB baseline memory usage
- **Startup**: < 5 seconds to ready state

Performance can be tuned via configuration:
- Batch size for Kafka polling
- Transaction commit interval
- State store flush frequency
- Connection pool sizes

## Migration from Java BaseAgent

If migrating from Java BaseAgent:

1. **Configuration**: Most configuration maps directly
2. **Message Format**: Same Envelope/Header structure
3. **State Storage**: Compatible state store keys
4. **Kafka Topics**: No changes needed
5. **Object Storage**: Same S3 paths and formats

Key differences:
- Async/await instead of threads
- Context managers for lifecycle
- Type hints instead of static typing
- Protocol classes instead of interfaces

## Troubleshooting

### Common Issues

1. **Kafka Connection Failed**
   - Check `kafka_brokers` configuration
   - Verify Kafka is running
   - Check network connectivity

2. **State Store Connection Failed**
   - For Redis: verify `redis_url`
   - Falls back to in-memory if Redis unavailable

3. **Health Check Not Responding**
   - Check `health_check_port` configuration
   - Verify port is not in use
   - Set to 0 to disable

4. **Message Processing Slow**
   - Check `kafka_max_poll_records`
   - Monitor transaction commit frequency
   - Review async operation bottlenecks

### Debug Logging

Enable debug logging:

```python
import logging
logging.basicConfig(level=logging.DEBUG)

# Or set environment variable
export LOG_LEVEL=DEBUG
```

## Contributing

Follow the TDD process documented in `python-base-agent-implementation-plan.md`:

1. Create Linear ticket
2. Create feature branch
3. Write tests FIRST
4. Implement to pass tests
5. Update documentation
6. Submit PR

## License

Proprietary - AI First Technologies LLC

## References

- [Java BaseAgent Topology](../base-agent-topology.md)
- [Implementation Plan](../python-base-agent-implementation-plan.md)
- [LLM Implementation Guide](../pythonAgent.md)