ai-agentic-baseagent

The Base Agent — reusable planner and orchestrator for multi-agent workflows.

---

Architecture overview

main.py
├── FastAPI  ──── REST API        /api/v1
├── A2A app  ──── Agent-to-Agent  /  (AgentCard + JSON-RPC)
└── FastMCP  ──── MCP tools       /mcp  (SSE transport)

app/
├── agent/                      # ADK agent — the purely agentic core
│   ├── base_agent.py           # LlmAgent instance (single source of truth)
│   ├── instructions.py         # System prompt & agent description
│   └── tools/
│       └── planning_tools.py   # Plain callable tool functions used by the LLM
│
├── a2a/                        # Agent-to-Agent protocol (A2A SDK)
│   ├── server.py               # A2A route registration + AgentCard
│   ├── executor.py             # BaseAgentExecutor (A2A → ADK bridge)
│   ├── registry.py             # Remote sub-agent discovery & registry
│   └── client.py               # Async helper — call any registered sub-agent by ID
│
├── mcp/                        # Model Context Protocol server (FastMCP)
│   └── mcp_server.py           # MCP tools: get_agent_info, list_registered_agents, run_production_plan
│
├── api/v1/                     # REST API (FastAPI)
│   ├── controllers/            # Route handlers (health, agent)
│   ├── services/               # ADK runner wrapper for REST consumers
│   ├── repositories/           # DB access (SQLAlchemy) — add per-entity files here
│   ├── schemas/                # Pydantic request / response models
│   └── dependencies/           # FastAPI Depends factories (auth, DB session, …)
│
├── cloud/                      # Cloud service clients
│   ├── s3.py                   # AWS S3 — upload, download, presigned URLs
│   └── secrets.py              # AWS Secrets Manager — cached secret retrieval
│
├── core/
│   ├── settings.py             # pydantic-settings (12-factor, all env-driven)
│   ├── logger.py               # Centralized logging with level-filter support
│   └── llm.py                  # LiteLLM abstraction — swap providers via .env
│
├── db/
│   ├── database.py             # Engine singleton, SessionLocal, init_db
│   └── models/                 # SQLAlchemy ORM models (add per-entity files)
│
├── common/
│   └── constants.py            # Enums, shared constants (TaskStatus, SubAgent, …)
│
└── di.py                       # DI container — all singleton services

---

Quick start

# 1. Clone & enter
cd ai-agentic-baseagent

# 2. Create and activate virtual environment
python3 -m venv venv && source venv/bin/activate

# 3. Install dependencies
pip install -r requirements.txt

# 4. Configure environment
cp .env.sample .env
# Edit .env — set GOOGLE_API_KEY (or LITELLM_* for proxy), DATABASE_URL, S3_BUCKET_NAME, …

# 5. Run database migrations
alembic upgrade head

# 6. Start the server
python main.py
# or: uvicorn main:app --host 0.0.0.0 --port 8080 --reload

---

Local Docker stack (local only)

This stack is for local development only and simulates:

• AWS RDS (MySQL 8)
• AWS S3 (via LocalStack)
• AWS SQS (via LocalStack)

Langfuse is not part of this local Docker stack. Use WPP-hosted Langfuse credentials provided by TL in your local .env.

Start

# from repo root
docker compose up -d --build

Check status

docker compose ps
docker compose logs -f localstack

Verify S3 bucket and SQS queue

docker exec produceragent-localstack awslocal s3api list-buckets && docker exec produceragent-localstack awslocal sqs list-queues

Expected resources:

• S3 bucket: wpp-briefs-local
• SQS queue: estimation-jobs

Verify MySQL

docker exec -it produceragent-db mysql -uroot -prootpassword -e "SHOW DATABASES;"

Stop

docker compose down

Reset (remove volumes)

docker compose down -v

---

Endpoints

Interface	Method	Path	Description
REST	GET	/api/v1/health	Liveness / readiness probe
REST	POST	/api/v1/agent/run	Submit a production request
REST	GET	/api/v1/agent/info	Agent metadata & registered sub-agents
A2A	GET	/.well-known/agent-card.json	AgentCard (agent discovery)
A2A	POST	/rpc	JSON-RPC task handler
MCP	GET	/mcp	FastMCP SSE endpoint
Docs	GET	/docs	Swagger UI
Docs	GET	/redoc	ReDoc UI

---

Database URL resolution

The DB DSN is resolved at startup using this priority chain (first match wins):

Priority	Mechanism	When to use
1	DATABASE_URL env var	Docker Compose / local dev / CI
2	DB_SECRET_NAME → AWS Secrets Manager	Production on AWS (ECS, Lambda, EC2)
3	SERVER_CONFIG_PATH Python file	On-prem / shared-hosting environments
4	DB_HOST + DB_PORT + DB_USERNAME + DB_PASSWORD	Local dev fallback

Secret format for AWS Secrets Manager (JSON):

{"host": "rds.example.com", "port": "3306", "username": "app", "password": "s3cr3t", "dbname": "prod"}

---

LLM configuration

The agent uses LiteLLM for provider-agnostic model access. To swap providers, change LITELLM_MODEL in .env — no code changes needed.

# Direct provider examples:
LITELLM_MODEL=gemini/gemini-2.0-flash      # Google Gemini
LITELLM_MODEL=openai/gpt-4o                # OpenAI
LITELLM_MODEL=anthropic/claude-3-5-sonnet  # Anthropic

# Via self-hosted LiteLLM proxy:
LITELLM_BASE_URL=https://your-proxy.example.com
LITELLM_VIRTUAL_KEY=sk-vk-xxxxxx
LITELLM_MODEL=openai/my-deployment-name

The abstraction lives in app/core/llm.py. To swap to a different ADK model backend (e.g. native Gemini, Vertex AI), edit only that file — the agent definition in app/agent/base_agent.py is unchanged.

API authentication

The orchestrator supports three auth modes via AUTH_MODE:

• secret (default): validate X-Secret-Key
• bearer: validate Authorization: Bearer <token>
• hybrid: accept either style (useful during integration migrations)

Configuration:

AUTH_MODE=secret
API_SECRET_KEY=change-me
# API_BEARER_TOKEN=change-me

Observability (Langfuse)

Langfuse tracing is integrated for all execution surfaces:

• REST: POST /api/v1/agent/run
• MCP: run_production_plan
• A2A: /rpc task execution

Configuration:

LANGFUSE_ENABLED=true
LANGFUSE_PUBLIC_KEY=pk-lf-xxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxx
LANGFUSE_HOST=https://cloud.langfuse.com
LANGFUSE_RELEASE=1.0.0
LANGFUSE_CAPTURE_INPUT=true
LANGFUSE_CAPTURE_OUTPUT=true
LANGFUSE_TAGS=base-agent,orchestrator
LANGFUSE_SAMPLE_RATE=1.0

Production notes:

• Keep LANGFUSE_* secrets out of source control.
• Use secret manager injection for runtime credentials.
• Reduce LANGFUSE_SAMPLE_RATE (for example 0.2) if needed for cost control.

Built-in test tools

The Base Agent includes deterministic dummy tools for local integration checks:

• ping() → returns {"status": "ok", "tool": "ping"}
• echo_payload(payload) → returns the same payload for contract testing
• simulate_subagent_result(task_id, assigned_agent, succeed=True) → returns a simulated task result with completed or failed status

These tools are defined in app/agent/tools/planning_tools.py and auto-registered via __all__.

Use them to validate ADK tool invocation, argument passing, and response handling before integrating real specialist sub-agents.

---

AWS S3

In production, attach an IAM role to your compute resource (ECS task role, EC2 instance profile, Lambda execution role) with s3:GetObject, s3:PutObject, etc. Do not set AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY in production.

In local dev, either:

• Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in .env, or
• Use LocalStack with S3_ENDPOINT_URL=http://localhost:4566.

S3 utilities: app/cloud/s3.py → S3Service.

---

Adding a sub-agent

Follow these four steps each time a new specialist agent is added:

1. Register the agent ID in the enum (app/common/constants.py):

class SubAgent(str, Enum):
    ESTIMATION = "estimation-agent"
    MY_NEW_AGENT = "my-new-agent"   # ← add here

2. Add a delegation tool in app/agent/tools/planning_tools.py:

def call_my_new_agent(input_url: str, job_id: str = "") -> dict[str, Any]:
    """
    Delegate a <describe task> to MyNewAgent via A2A.

    Args:
        input_url: Presigned URL of the input artifact.
        job_id:    Identifier for tracing.
    """
    payload = {"input_url": input_url, "job_id": job_id}
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            import concurrent.futures
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
                future = pool.submit(asyncio.run, call_agent(SubAgent.MY_NEW_AGENT, payload))
                response_text = future.result()
        else:
            response_text = loop.run_until_complete(call_agent(SubAgent.MY_NEW_AGENT, payload))
    except Exception as exc:
        return {"error": str(exc), "task": "call_my_new_agent"}
    try:
        return json.loads(response_text)
    except json.JSONDecodeError:
        return {"result": response_text}

3. Export the tool by adding it to __all__:

__all__ = [
    ...
    "call_my_new_agent",
]

4. Register the AgentCard URL in .env:

SUB_AGENT_CARDS=[
  {"id": "estimation-agent", "card_url": "http://estimation:4014/.well-known/agent.json"},
  {"id": "my-new-agent",     "card_url": "http://my-new-agent:PORT/.well-known/agent.json"}
]

The AgentRegistry fetches every card at startup. If a card URL is unreachable, startup logs a warning but continues — the affected tool will return a clear error at call time.

Current sub-agent roster

ID	Enum	Port (default)	Pathway
estimation-agent	SubAgent.ESTIMATION	4014	estimation

---

Key technologies

Technology	Role
Google ADK	Agent runtime (LlmAgent, Runner, InMemorySessionService)
LiteLLM	Provider-agnostic LLM routing (swap models via config)
A2A SDK	Inter-agent communication (AgentCard, BaseAgentExecutor)
FastMCP	MCP server with SSE transport
FastAPI	REST API framework
SQLAlchemy + PyMySQL	ORM + MySQL driver
Alembic	Database schema migrations
Pydantic v2	Request/response validation & settings management
boto3	AWS SDK — S3 + Secrets Manager

python3 -m venv .venv && source .venv/bin/activate

Install the packages

pip3 install -r requirements.txt

Create .env and replace the necessary keys

cp .env.sample .env

uvicorn main:app --reload --port 8002