AgentBharat

Turn AI agents into microservices. Deploy, compose, and scale.

One decorator to make any AI agent a production-ready microservice with REST API, health checks, and auto-generated docs.

Why AgentBharat?

Building AI agents is easy. Deploying them as reliable, scalable services is hard. AgentBharat bridges that gap:

• One decorator — @agent_service turns any function into a FastAPI microservice
• Auto-generated API — REST endpoints, OpenAPI docs, health checks out of the box
• Pipeline composition — Chain agents with >> operator: researcher >> writer >> reviewer
• Service discovery — Agents register themselves and find each other
• Message bus — Async communication between agents (in-memory or Redis)
• Typed contracts — Pydantic-based input/output validation between services
• Built-in dashboard — Web UI to monitor your agent mesh
• Scale independently — Each agent runs as its own process, scale what you need

Installation

pip install AgentBharat

# With Redis support for production message bus
pip install AgentBharat[redis]

Quick Start

1. Create an Agent Service

from AgentBharat import agent_service

@agent_service(name="researcher", port=8001)
async def research(topic: str) -> dict:
    """Research agent — runs as its own microservice."""
    # Your LLM logic here
    return {"findings": f"Key insights about {topic}...", "sources": 3}

# Start the service
research.serve()
# → API running at http://localhost:8001
# → Docs at http://localhost:8001/docs

2. Call It from Anywhere

from AgentBharat import AgentClient

client = AgentClient("http://localhost:8001")
result = await client.run(topic="quantum computing")
print(result.output)
# → {"findings": "Key insights about quantum computing...", "sources": 3}

3. Chain Agents into Pipelines

from AgentBharat import agent_service

@agent_service(name="researcher", port=8001)
async def researcher(topic: str) -> dict:
    return {"findings": f"Research on {topic}..."}

@agent_service(name="writer", port=8002)
async def writer(findings: str) -> dict:
    return {"article": f"Blog post based on: {findings}"}

@agent_service(name="reviewer", port=8003)
async def reviewer(article: str) -> str:
    return f"Reviewed and approved: {article[:100]}..."

# Compose with >> operator
pipeline = researcher >> writer >> reviewer
result = await pipeline.run(topic="AI agents")
print(result.output)

4. Service Discovery

from AgentBharat import ServiceRegistry

registry = ServiceRegistry()

# Services register themselves
registry.register("researcher", "http://localhost:8001", port=8001, tags=["research"])
registry.register("writer", "http://localhost:8002", port=8002, tags=["content"])

# Discover by name
service = registry.get("researcher")
print(service.url)  # → http://localhost:8001

# Discover by tag
content_services = registry.find_by_tag("content")

5. Message Bus

from AgentBharat.bus import InMemoryBus, Message

bus = InMemoryBus()

# Subscribe to events
async def on_research_done(msg: Message):
    print(f"Research complete: {msg.payload}")

await bus.subscribe("research-done", on_research_done)

# Publish events
await bus.publish("research-done", Message.create(
    source="researcher",
    target="writer",
    payload={"findings": "..."},
))

6. Dashboard

from AgentBharat.dashboard import build_dashboard_app
from AgentBharat import ServiceRegistry
import uvicorn

registry = ServiceRegistry()
registry.register("researcher", "http://localhost:8001", port=8001)
registry.register("writer", "http://localhost:8002", port=8002)

app = build_dashboard_app(registry, port=9000)
uvicorn.run(app, port=9000)
# → Dashboard at http://localhost:9000

7. Typed Contracts

from AgentBharat import agent_service, InputSchema, OutputSchema, Contract

class ResearchInput(InputSchema):
    topic: str
    max_sources: int = 5

class ResearchOutput(OutputSchema):
    findings: str
    sources: list[str]

contract = Contract.from_types(
    name="research",
    input_type=ResearchInput,
    output_type=ResearchOutput,
)

# Contract validates data at runtime
contract.validate_input({"topic": "AI"})  # OK
contract.validate_input({})  # Raises ValueError: Missing required field 'topic'

Architecture

AgentBharat/
├── service.py          # @agent_service decorator & FastAPI service
├── client.py           # HTTP client for calling services
├── contracts.py        # Typed input/output contracts
├── dashboard.py        # Web UI for monitoring
├── router/
│   └── registry.py     # Service discovery & registration
├── bus/
│   ├── base.py         # Abstract message bus
│   ├── memory.py       # In-memory bus (dev/testing)
│   └── redis_bus.py    # Redis bus (production)
└── pipeline/
    └── builder.py      # Pipeline composition with >> operator

Roadmap

• gRPC support alongside REST
• Docker/Kubernetes deployment helpers
• Circuit breaker pattern for resilience
• Agent authentication and API keys
• Webhook triggers and cron scheduling
• Prometheus metrics export
• CLI tool for managing agent services

Contributing

git clone https://github.com/abhishekjain/AgentBharat.git
cd AgentBharat
pip install -e ".[dev]"
pytest

License

MIT License — see LICENSE for details.