Develop against the API

Overview

Agent Barn has two FastAPI composition roots.

Product API                       Runtime Ingest API
http://localhost:8000/api/v1      http://localhost:8001/ingest/v1

The Product API serves authenticated users, Organization members, and Platform Administrators. The Ingest API accepts telemetry emitted by running Agents.

They share application infrastructure, but they do not share an authentication contract. A user access token is for the Product API. A per-Agent Ingest key is for that Agent’s Ingest endpoint.

Dependency direction

The normal backend dependency direction is:

HTTP request
    │
    ▼
FastAPI route
    │
    ▼
Service
    ├──────────────► Infrastructure adapter
    │
    ▼
Repository
    │
    ▼
PostgreSQL

Routes stay thin. Services make business and authorization decisions. Repositories own queries and persistence. Infrastructure adapters own external concerns such as Kubernetes, Slack, email, LiteLLM, OpenRouter, and encryption.

Choose the API boundary

Select the boundary before adding a route or writing a client.

Product API route families

• /auth
• /organizations
• /organizations/{organization_id}/members
• /organizations/{organization_id}/agents
• /organizations/{organization_id}/templates
• /organizations/{organization_id}/skills
• /organizations/{organization_id}/shared-credentials
• /organizations/{organization_id}/costs
• /integrations
• /platform/users
• /platform/organizations
• /platform/templates
• /platform/skills
• /platform/stats
• /platform/event-deliveries

The Ingest surface

The Ingest surface is intentionally narrow:

POST /ingest/v1/agents/{agent_id}/events

Provider webhooks are also separate trust boundaries. Microsoft Teams messages arrive through /api/v1/webhooks/teams/{agent_id}/messages and are handled using provider-specific verification and relay behavior rather than the normal user-session dependency.

Run the API

make setup
make run

make db-up
make migrate
make dev-api

make dev-api starts:

• Product API on port 8000
• Ingest API on port 8001

It also supplies the host-facing Ingest URL used by Agents running in the local Kubernetes cluster.

Verify the API and its database connection

curl --fail-with-body \
  http://localhost:8000/api/v1/health

A healthy response is:

{
  "status": "ok",
  "db": "connected"
}

A 503 response means the API process is reachable but cannot query PostgreSQL.

Inspect the contract

FastAPI publishes interactive documentation and the machine-readable OpenAPI schema for each mounted application.

Use the OpenAPI document to confirm:

• HTTP method and path
• Required path and query parameters
• Request content type
• Request DTO
• Response DTO
• Validation constraints
• Status codes
• Whether bearer authentication is required

Treat the response DTO as the external contract. A database model may contain encrypted, internal, or operational fields that must never appear in an API response.

Contract conventions

A paginated response has this general shape:

{
  "page": 1,
  "page_size": 15,
  "total": 2,
  "items": [
    {
      "id": "00000000-0000-0000-0000-000000000000"
    }
  ]
}

Do not infer writable fields from a read response. Inspect the endpoint’s *Create or *Update schema.

Authenticate requests

Product API calls normally use a user-session bearer access token.

1. 1

   Log in

   The login endpoint accepts OAuth2 form data, not a JSON body. Supply the user’s email through the form field named username.

   Log in Copy

   curl --request POST \
     --url http://localhost:8000/api/v1/auth/login \
     --header 'Content-Type: application/x-www-form-urlencoded' \
     --data-urlencode '[email protected]' \
     --data-urlencode 'password=replace-with-your-password'

   The response contains an access token, refresh token, and bearer token type:

   Token response Copy

   {
     "access_token": "replace-with-returned-access-token",
     "refresh_token": "replace-with-returned-refresh-token",
     "token_type": "bearer"
   }

   The server also writes the refresh token to an HttpOnly cookie. Local and test environments use local-compatible cookie settings; deployed environments use secure cookie settings.

2. 2

   Send the access token

   Store the returned access token in a temporary environment variable:

   Store the token Copy

   export AGENT_BARN_ACCESS_TOKEN='replace-with-returned-access-token'

   Confirm the authenticated user:

   Current user Copy

   curl --fail-with-body \
     http://localhost:8000/api/v1/auth/me \
     --header "Authorization: Bearer ${AGENT_BARN_ACCESS_TOKEN}"

   All protected Product API requests use the same header:

   Authorization header Copy

   Authorization: Bearer <access-token>

3. 3

   Refresh the session

   A browser client can refresh using the HttpOnly cookie:

   Refresh with the cookie Copy

   curl --request POST \
     --url http://localhost:8000/api/v1/auth/refresh \
     --cookie 'refresh_token=replace-with-refresh-token'

   A non-browser client can supply the refresh token in the request body:

   Refresh with a request body Copy

   curl --request POST \
     --url http://localhost:8000/api/v1/auth/refresh \
     --header 'Content-Type: application/json' \
     --data '{
       "refresh_token": "replace-with-refresh-token"
     }'

   Refresh tokens are rotated. Replace the stored access and refresh tokens with the values returned by the refresh request.

Scope requests

Authentication establishes who the caller is. It does not automatically grant access to every Organization or Agent.

Most Product API requests resolve access through this sequence:

1. 1
   Authenticated user

   A valid, unexpired, correctly typed bearer access token identifies the caller.

2. 2
   Active Organization

   The authentication dependency reads organization_id from the route and resolves a real persisted Membership.

3. 3
   Organization Permission

   The service requires the specific Permission its operation needs, resolved from the caller’s Organization Role.

4. 4
   Agent visibility

   Repository queries conceal Agents and subordinate resources the caller cannot see, before counting and pagination.

5. 5
   Agent action Permission

   Effective Agent permissions are resolved again on the server for every mutation.

Organization scope

Organization routes carry the Organization UUID in the URL:

/api/v1/organizations/{organization_id}/...

For example:

export AGENT_BARN_ORGANIZATION_ID='replace-with-organization-id'

curl --fail-with-body \
  "http://localhost:8000/api/v1/organizations/${AGENT_BARN_ORGANIZATION_ID}/agents?page=1&page_size=15" \
  --header "Authorization: Bearer ${AGENT_BARN_ACCESS_TOKEN}"

The authentication dependency reads organization_id from the route and resolves a real persisted Membership for the user.

Routes without an organization_id path parameter do not have an active Organization. Examples include login, the current-user endpoint, Organization creation, and Platform Administrator routes.

Platform scope

Platform administration routes live under:

/api/v1/platform/...

These routes use the Platform Administrator dependency and do not synthesize an active Organization.

Agent scope

Agent authorization adds another layer beneath Organization Membership.

Organization Owners and Administrators receive the documented implicit Agent authority. Other members can receive access through:

• Explicit Agent Access
• An Agent Access Role
• Agent General Access, where applicable
• The union of effective Agent permissions

Agent and subordinate-resource queries must enforce visibility in the repository before counting or paginating results. This applies to:

• Conversations
• Tool Calls
• Activity
• Logs
• Costs
• Secrets
• Skills assigned to an Agent
• Agent configuration

The API may include allowed_actions in Agent read responses so clients can render appropriate controls. That field is a client hint, not authorization proof. The backend resolves permissions again for every mutation.

HTTP security semantics

Tenant-sensitive resources generally return 404 when an identifier belongs to another Organization. This prevents the response from confirming that the resource exists.

Organization administration can return 403 when the caller is known but lacks Organization authority. Always inspect the endpoint’s tested contract instead of converting every authorization failure to the same status.

Follow the request lifecycle

A normal Product API request follows this path:

The business change and its event records must commit together. Event handling occurs after that transaction boundary.

The layer boundaries are deliberate. Each layer owns a distinct set of decisions.

Add an API capability

Start by reading the neighboring domain. New domains normally use:

api/domains/<domain>/
├── models.py
├── repository.py
├── service.py
└── routes.py

Add another file only when it owns a distinct responsibility such as parsing, building, provider behavior, or domain exceptions.

The workflow below usually touches:

• api/domains/<domain>/models.py
• api/domains/<domain>/repository.py
• api/domains/<domain>/service.py
• api/domains/<domain>/routes.py
• api/api_app.py
• api/migrations/versions/

1. 1

   Define the contract

   Create separate types for persistence and API input and output. Database models and API DTOs remain distinct types.

   Use the established suffixes:

   • *Create
   • *Update
   • *Read
   • *Filter

   Keep encrypted and internal fields out of read DTOs. Use default_factory for mutable defaults.

   For partial updates, distinguish an omitted field from a field explicitly set to null:

   Partial update Copy

   changes = update.model_dump(exclude_unset=True)

2. 2

   Add repository behavior

   Repositories own query composition.

   An Organization-scoped lookup must include its tenant boundary in the query. Do not fetch a globally addressed row and filter it in the service afterward.

   The Skill repository demonstrates the pattern:

   Tenant-scoped lookup Copy

   def get_by_id_for_org(self, skill_id: UUID, org_id: UUID) -> Skill | None:
       with Session(self.delegate.engine) as session:
           return session.exec(
               select(Skill).where(
                   col(Skill.id) == skill_id,
                   or_(
                       col(Skill.organization_id) == org_id,
                       col(Skill.organization_id).is_(None),
                   ),
               )
           ).first()

   The organization_id IS NULL branch is specific to globally available built-in Skills. Do not copy that allowance into a domain whose records must always belong to one Organization.

   For Agent or subordinate-resource collections, reuse the existing accessible-Agent query pattern. Visibility must affect both the result query and the count query.

3. 3

   Add service behavior

   Services derive the active Organization from CurrentUserContext, enforce permissions, and coordinate the work.

   A representative Organization permission check is:

   Organization permission Copy

   org_id = context.require_current_user_organization().organization_id
   
   self.permission_policy.require_organization(
       context,
       org_id,
       PermissionKey.SKILL_MANAGE,
   )

   For Agent resources, use the shared Agent authorization module rather than reproducing role checks:

   Agent action permission Copy

   agent = self.agent_authorization.require_action(
       context,
       agent_id,
       PermissionKey.AGENT_UPDATE,
   )

   Resolve grants on each request. Never trust a client-supplied role, allowed_actions value, or permission flag.

   Translate expected domain failures in the service:

   Domain failure	Status
   Invalid business input	400
   Missing or concealed record	404
   Visible but disallowed action	403
   Uniqueness or lifecycle conflict	409

4. 4

   Add a thin route

   A representative route follows this shape:

   Route pattern Copy

   skills_router = APIRouter(
       prefix="/organizations/{organization_id}/skills",
       tags=["skills"],
   )
   
   
   @skills_router.post(
       "",
       response_model=SkillSummaryRead,
       status_code=status.HTTP_201_CREATED,
   )
   def create_skill(
       data: SkillCreate,
       context: Annotated[
           CurrentUserContext,
           Depends(get_current_user()),
       ],
       service: Annotated[
           SkillService,
           Injected(SkillService),
       ],
   ):
       return service.create_skill(data, context)

   The route declares the contract and delegates the workflow. The authentication dependency resolves the Organization ID from the request path.

   Use the established status conventions:

   Operation	Status
   Read	200
   Update returning a body	200
   Create	201
   Delete or bodyless action	204

   Prefer 204 to custom success objects.

5. 5

   Use dependency injection

   Domain classes normally use injector:

   Injected service Copy

   @inject
   @singleton
   @dataclass
   class SkillService:
       repository: SkillRepository
       permission_policy: PermissionPolicy

   FastAPI routes receive services using Injected(...).

   Do not construct repositories, services, infrastructure clients, or database engines inside route handlers.

   Shared infrastructure providers are configured through AppModule and composed by create_injector(). Add a provider when construction requires configuration or a deliberately selected implementation.

6. 6

   Register the router

   Register a new Product API router in api/api_app.py using the existing composition pattern:

   Router registration Copy

   subapi.include_router(skills_router)

   Register runtime telemetry routes through api/ingest_app.py.

7. 7

   Synchronize consumers and documentation

   When an API contract changes, check every consumer. Update, as applicable:

   • Response and request DTOs
   • UI Zod schemas
   • UI query hooks and mutations
   • Mock responses
   • Playwright fixtures and expectations
   • Generated or external clients
   • Authoritative architecture or feature documentation

   If a change revises a domain invariant, authorization boundary, lifecycle state, or operational contract, update the corresponding document in the same application-repository change.

Preserve atomicity

PostgresRepositoryDelegate is appropriate for ordinary persistence. Its operations open and commit a database session per operation.

That means this service workflow is not automatically atomic:

repository.save(first_record)    → commit
repository.save(second_record)   → commit

If the second operation fails, the first may already be durable.

One transaction for business state and events

An event-producing mutation persists business state, an immutable Outbox Message, and the intended Event Deliveries in one transaction. The repository should own the session:

with Session(self.delegate.engine, expire_on_commit=False) as session:
    session.add(resource)
    session.flush()

    outbox_repository.stage(
        session=session,
        registry=EVENT_REGISTRY,
        event=event,
    )

    session.commit()

The outbox staging operation must use the caller’s session. It must not open or commit another session.

After the transaction commits, enqueue immediate delivery or allow reconciliation to process the pending Event Deliveries.

Do not pass database sessions into routes. Avoid holding a database transaction open across a slow external network call unless the domain explicitly requires that failure behavior.

Evolve the schema

A database contract change is more than a model edit. Update all affected surfaces:

• Database model
• Request DTO
• Response DTO
• Repository query
• PostgreSQL constraints or indexes
• Alembic migration
• Integration-test setup and assertions
• UI Zod schema when the web app consumes the contract
• Authoritative feature or architecture documentation when an invariant changes

Create a migration from the repository root, review it, and apply it:

make makemigrations
make migrate
make check-migrations

Review the generated migration under api/migrations/versions/, then confirm the migration graph has one head with make check-migrations.

Exercise the immediate downgrade when rollback is supported:

make rollback
make migrate

Check PostgreSQL-specific behavior explicitly:

• Foreign-key ownership
• ON DELETE behavior
• Nullability
• Uniqueness
• Indexes
• Check constraints
• Enum creation and removal
• Server defaults
• Migration order
• Data backfills for existing rows

Test the contract

Choose the lowest test layer that proves the behavior reliably, then add integration coverage when the HTTP, authentication, tenancy, database, or composition contract is part of the change.

How API integration tests run

API integration tests use:

• The real FastAPI application
• A migrated PostgreSQL database
• PostgreSQL Testcontainers
• Additive Injector overrides
• Reusable Given/When/Then steps
• PyHamcrest assertions

Test shape

A representative test uses Given/When/Then steps and PyHamcrest assertions:

def test_member_cannot_read_another_organizations_agent():
    with given([
        prepare_injector(),
        prepare_api_server(),
        create_test_client(),
        there_is_a_user(),
        there_is_an_access_token_for_user(),
        there_is_an_agent_in_another_organization(),
    ]) as context:
        with when("the member requests the other Organization's Agent"):
            response = context.client.get(
                context.agent_url,
                headers={
                    "Authorization": f"Bearer {context.access_token}",
                },
            )

            with then("the concealed Agent is not found"):
                assert_that(
                    response.status_code,
                    equal_to(status.HTTP_404_NOT_FOUND),
                )

Keep shared setup in the existing test support structure instead of embedding large arrangements in every test.

What to cover

• Happy path
• Unauthenticated request
• Missing Organization Membership
• Insufficient Organization Permission
• Missing Agent access
• Visible Agent with insufficient action Permission
• Cross-Organization identifier
• Important validation failures
• Not-found behavior
• State or uniqueness conflicts
• Migration behavior when the schema changes

Verify the change

Run the API checks from the repository root:

make check-api
make test-api
make check-migrations

make check-api runs:

• Ruff lint checks
• Ruff formatting checks
• Python type checking

Run Kubernetes integration only when the changed contract depends on Kubernetes behavior:

make test-api-k8s

If the API contract is consumed by the web app, also run:

make lint-ui
make check-ui
make test-ui

Troubleshooting

Next steps