Overview

Overview¶

concierge/cloud_agent is an asynchronous task dispatch application built with clean architecture (DDD layered structure). It receives tasks via a REST API, enqueues them in a background job queue, routes them to the appropriate agent, and returns results.

flowchart LR
    Client[REST Client] --> Web[FastAPI Routes]
    Web --> UC[Application Use Cases]
    CLI[Typer CLI / Worker] --> UC
    UC --> Domain[Domain Entities]
    UC --> Repo[TaskRepository]
    UC --> Queue[TaskQueue]
    UC --> Registry[AgentRegistry]
    Registry --> Echo[EchoAgent]
    Registry --> LG["LangGraphAgent\n(langgraph)"]
    Registry --> GCE[GitHubCopilotSdkAgent]
    Registry --> MAF["MicrosoftAgentFrameworkAgent\n(microsoft-agent-framework)"]

Agent Extension Point¶

Agents are defined in the shared concierge/agents/ package. Each agent implements the Agent Protocol from the shared application layer:

class Agent(Protocol):
    # Plain attribute (satisfied by both ``ClassVar[str]`` on single-purpose
    # agents and instance attributes on configurable agents).
    agent_type: str
    async def handle(self, request: AgentRequest) -> AgentResponse: ...

The infrastructure layer is the only place that may import LangChain / LangGraph. The domain and application layers must remain framework-free (enforced by tests/agents/test_architecture.py and import-linter contracts).

classDiagram
    class Agent {
        <<Protocol>>
        +agent_type: str
        +handle(request) AgentResponse
    }
    class EchoAgent {
        +agent_type = "echo"
        +handle(request) AgentResponse
    }
    class LangGraphAgent {
        +agent_type (instance)
        +tool_builders
        +handle(request) AgentResponse
        -_build_agent(side_outputs)
    }
    class GitHubCopilotSdkAgent {
        +agent_type = "github-copilot-sdk"
        +handle(request) AgentResponse
    }
    class MicrosoftAgentFrameworkAgent {
        +agent_type (instance)
        +tool_builders
        +handle(request) AgentResponse
        -_build_agent(side_outputs)
    }
    Agent <|.. EchoAgent
    Agent <|.. LangGraphAgent
    Agent <|.. GitHubCopilotSdkAgent
    Agent <|.. MicrosoftAgentFrameworkAgent

Key Design Principles¶

Queue-agnostic abstraction — ships with InMemory (local dev) and AzureStorageQueue backends; switch via CLOUD_AGENT_QUEUE_BACKEND.
Agent I/O standardised — every agent receives AgentRequest and returns AgentResponse (Pydantic schemas from concierge.agents). The AgentRegistry maps agent_type strings to concrete Agent implementations.
Runtime-agnostic workers — the worker loop runs as a local CLI process today; the same Agent interface can be reused on Azure Functions in the future.
Dead Letter Queue (DLQ) — tasks that exceed max_retries are moved to a DLQ automatically.

Directory Layout¶

concierge/cloud_agent/
  domain/
    entities.py        # Task dataclass with state-machine transitions
    value_objects.py   # TaskStatus enum + allowed transitions
    exceptions.py      # Domain-specific exceptions
  application/
    agents.py          # Agent Protocol, TaskInput/Output, AgentRegistry
    queues.py          # TaskQueue Protocol + QueueMessage schema
    repositories.py    # TaskRepository Protocol
    use_cases.py       # DispatchTask, GetTask, ListTasks, CancelTask, …
  infrastructure/
    persistence/       # InMemoryTaskRepository, SqlAlchemyTaskRepository
    queue/             # InMemoryTaskQueue, AzureStorageQueueTaskQueue
    web/               # FastAPI app, routes, schemas, exception handlers
    cli/               # Typer CLI app, worker loop

Quick Start¶

# Start the REST API (in-memory backend by default)
uv run cloud-agent-web

# Run the worker (separate terminal)
uv run cloud-agent-cli worker

# Dispatch a task
uv run cloud-agent-cli task dispatch --agent-type echo --payload '{"message": "hello"}'

# List registered agents
uv run cloud-agent-cli agents

Running the LangGraph Agent¶

The langgraph preset is the reference setup for integrating LangChain / LangGraph agents with the cloud_agent task pipeline. It is built on the unified LangGraphAgent class, which uses langchain.agents.create_agent with the echo and generate_image_tool tool builders and an Azure-hosted chat model resolved through init_chat_model. The LLM picks the appropriate tool based on the user's request. Adding additional tools means extending the tool_builders list — not creating a new agent class.

Prerequisites¶

Azure AI Foundry (or Azure OpenAI) deployment reachable via the model string in AGENTS_LANGGRAPH_MODEL (default azure_ai:gpt-5).
A principal that DefaultAzureCredential can resolve — typically az login for local development, or a managed identity in Azure.
The signed-in principal must have permission to call the Foundry deployment (e.g. Azure AI Developer role).

Minimal `.env`¶

The fastest setup uses both in-memory backends. The API and the worker must run in the same Python process to share the queue / repository, so this mode is only useful for embedded smoke tests. For a realistic split (separate cloud-agent-cli worker and cloud-agent-cli task dispatch processes), switch to postgres + azure-storage-queue.

# .env — split-process setup
CLOUD_AGENT_REPOSITORY_BACKEND=postgres
CLOUD_AGENT_QUEUE_BACKEND=azure-storage-queue
CLOUD_AGENT_AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net
AGENTS_LANGGRAPH_MODEL=azure_ai:gpt-5

POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_USER=concierge
POSTGRES_PASSWORD=concierge
POSTGRES_DB=concierge

Step-by-step¶

# 1. Authenticate so DefaultAzureCredential can mint tokens
az login

# 2. Start dependencies (only required for postgres / azure-storage-queue)
docker compose up -d postgres

# 3. Confirm the agent is registered
uv run cloud-agent-cli agents
# → ["echo", "langgraph", "github-copilot-sdk", "microsoft-agent-framework"]

# 4. Start the worker (terminal 1)
uv run cloud-agent-cli worker

# 5. Dispatch a task (terminal 2)
uv run cloud-agent-cli task dispatch \
  --agent-type langgraph \
  --payload '{"message": "Hello LangGraph"}'

# 6. Poll for the result using the task id printed above
uv run cloud-agent-cli task get <task-id>

Payload contract¶

Field	Type	Required	Notes
`message`	`string`	Yes	Non-empty. Forwarded verbatim to the LLM; the agent fails with `payload.message is required` otherwise.

Result shape¶

A successful task stores the following object under result:

{
  "message": "Hello LangGraph",
  "reply": "<final assistant message>",
  "tool_calls": [
    {"name": "echo", "args": {"text": "Hello LangGraph"}}
  ],
  "model": "azure_ai:gpt-5"
}

reply is the last AIMessage.content produced by the graph, and tool_calls lists every (name, args) pair the model emitted while processing the task. The model field echoes the configured AGENTS_LANGGRAPH_MODEL.

Customising the agent¶

AGENTS_LANGGRAPH_MODEL — swap the underlying chat model (e.g. azure_ai:gpt-4o-mini).
AGENTS_LANGGRAPH_SYSTEM_PROMPT — replace the built-in system prompt to change behaviour without writing code.
To add a new tool variant, write a tool builder under concierge/agents/infrastructure/tools/ and register an extra LangGraphAgent(...) preset in concierge/agents/infrastructure/registry_factory.py.

Troubleshooting¶

Symptom	Likely cause
Task stuck in `QUEUED`	Worker not running, or `memory` backend used across separate processes. Start `cloud-agent-cli worker` or switch to `postgres` + `azure-storage-queue`.
`status=failed`, error mentions credentials	`DefaultAzureCredential` could not resolve a principal. Run `az login` or configure a managed identity.
`status=failed`, `payload.message is required`	The dispatched payload is missing `message` or it is an empty / whitespace-only string.
403 from the model deployment	The principal is missing the Azure AI Developer role on the Foundry project.

Task Lifecycle¶

QUEUED → RUNNING → SUCCEEDED
                 → FAILED → (retry) → QUEUED
                          → (max retries) → DEAD_LETTER
       → CANCELLED

Configuration¶

All Cloud Agent settings are aggregated in concierge.settings.CloudAgentSettings and read from environment variables with the CLOUD_AGENT_ prefix (see .env.template). The rest of the codebase never touches os.environ directly — both the REST API (cloud-agent-web) and the worker / dispatcher CLI (cloud-agent-cli) share the exact same configuration object.

Settings reference¶

Variable	Default	Description
`CLOUD_AGENT_REPOSITORY_BACKEND`	`memory`	Task persistence backend: `memory` / `postgres` / `azure-postgres`.
`CLOUD_AGENT_TABLE_NAME`	`cloud_agent_tasks`	Table name used by the SQL backends.
`CLOUD_AGENT_QUEUE_BACKEND`	`memory`	Job queue backend: `memory` / `azure-storage-queue`.
`CLOUD_AGENT_QUEUE_NAME`	`cloud-agent-tasks`	Main task queue name (Azure Storage Queue resource name).
`CLOUD_AGENT_DLQ_NAME`	`cloud-agent-dlq`	Dead Letter Queue name (created automatically on first use).
`CLOUD_AGENT_AZURE_STORAGE_ACCOUNT_URL`	(empty)	Required when `CLOUD_AGENT_QUEUE_BACKEND=azure-storage-queue`. Queue service endpoint (e.g. `https://<account>.queue.core.windows.net`). Authentication is performed only via Microsoft Entra ID (`DefaultAzureCredential`); connection strings / account keys are not supported.
`CLOUD_AGENT_VISIBILITY_TIMEOUT_SECONDS`	`60`	How long a dequeued message is hidden from other workers while a task is being processed. Should comfortably exceed the worst-case agent execution time.
`CLOUD_AGENT_MAX_RETRIES`	`3`	Default retry budget injected into `DispatchTaskUseCase`. A task is moved to the DLQ when `retry_count > max_retries`. Can be overridden per dispatch via the API / CLI.
`CLOUD_AGENT_WORKER_CONCURRENCY`	`1`	Reserved for future concurrent processing per worker. The current loop processes one task at a time; scale horizontally instead.
`CLOUD_AGENT_POLL_INTERVAL_SECONDS`	`1.0`	How long the worker sleeps after an empty `dequeue()` before polling again.

LangGraph agent settings (AGENTS_LANGGRAPH_MODEL, AGENTS_LANGGRAPH_SYSTEM_PROMPT) are now managed in the shared agent runtime. See Shared Agent Runtime.

Repository backend selection¶

The repository persists Task aggregates and is consumed by both cloud-agent-web and the worker. The choice is made by CLOUD_AGENT_REPOSITORY_BACKEND:

Value	Enum member	When to use	Schema init
`memory` (default)	`CloudAgentRepositoryBackend.MEMORY`	Local smoke tests. Data is lost on restart and not shared across processes — the worker and API must run in the same process to see each other’s tasks.	Not needed
`postgres`	`CloudAgentRepositoryBackend.POSTGRES`	Local Docker Compose PostgreSQL using the `POSTGRES_*` variables. Tables are created lazily by `SqlAlchemyTaskRepository` on first use.	Auto
`azure-postgres`	`CloudAgentRepositoryBackend.AZURE_POSTGRES`	Azure Database for PostgreSQL Flexible Server using the `AZURE_*` variables. Supports Microsoft Entra ID auth via `AZURE_USE_ENTRA_AUTH=true`.	Auto

memory and multi-process setups

Because InMemoryTaskRepository stores state in a Python dict, running cloud-agent-web and cloud-agent-cli worker in separate terminals with memory mode will produce two completely independent task stores. Switch to postgres (or azure-postgres) whenever you split processes.

Queue backend selection¶

Value	When to use	Required variables
`memory` (default)	Local smoke tests. Uses an in-process `asyncio.Queue`. Only useful when the API and the worker live in the same Python process.	—
`azure-storage-queue`	Production-grade durable queue. The main queue and DLQ are auto-created on startup. Auth is Entra ID only via `DefaultAzureCredential`.	`CLOUD_AGENT_AZURE_STORAGE_ACCOUNT_URL`

The factory raises ValueError if azure-storage-queue is selected without an account URL. Authentication is performed exclusively via Microsoft Entra ID using DefaultAzureCredential — the calling principal must hold the Storage Queue Data Contributor role (or equivalent custom RBAC) on the storage account. Visibility timeouts and DLQ routing are driven by CLOUD_AGENT_VISIBILITY_TIMEOUT_SECONDS and CLOUD_AGENT_DLQ_NAME.

Example: local development (memory only)¶

# .env (minimum)
CLOUD_AGENT_REPOSITORY_BACKEND=memory
CLOUD_AGENT_QUEUE_BACKEND=memory

Run the API and worker in the same process group so they share the in-memory store and queue. The simplest way is to keep them in one terminal each but understand they will use independent in-memory state — use this mode only for unit-style end-to-end checks.

Example: PostgreSQL + Azure Storage Queue¶

# .env
CLOUD_AGENT_REPOSITORY_BACKEND=postgres
CLOUD_AGENT_QUEUE_BACKEND=azure-storage-queue
CLOUD_AGENT_AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net
CLOUD_AGENT_QUEUE_NAME=cloud-agent-tasks
CLOUD_AGENT_DLQ_NAME=cloud-agent-dlq
CLOUD_AGENT_VISIBILITY_TIMEOUT_SECONDS=120
CLOUD_AGENT_MAX_RETRIES=5

# Local Postgres connection (shared with the Chat / Todo apps)
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_USER=concierge
POSTGRES_PASSWORD=concierge
POSTGRES_DB=concierge

docker compose up -d postgres
az login                       # so DefaultAzureCredential can mint a token
uv run cloud-agent-web        # terminal 1
uv run cloud-agent-cli worker # terminal 2

Grant the signed-in principal (or managed identity) the Storage Queue Data Contributor role on the target storage account before starting the services.

Example: Azure Database for PostgreSQL (Entra ID) + Azure Storage Queue¶

# .env
CLOUD_AGENT_REPOSITORY_BACKEND=azure-postgres
CLOUD_AGENT_QUEUE_BACKEND=azure-storage-queue
CLOUD_AGENT_AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net

AZURE_DBHOST=<server-name>.postgres.database.azure.com
AZURE_DBNAME=postgres
AZURE_USE_ENTRA_AUTH=true
AZURE_DBUSER=<entra-principal>

Ensure DefaultAzureCredential can mint a token (e.g. az login or a managed identity) before starting the API / worker. The same principal must have the Storage Queue Data Contributor role on the storage account in addition to the PostgreSQL role used for AZURE_DBUSER.