Embedding Agents in Existing Systems

Production environments tend to accumulate legacy systems. You have a Django app that handles customer orders. You have a Jenkins pipeline that builds and deploys your services. You have a PostgreSQL database that tracks inventory. You have a Slack workspace where your team coordinates. How do we add agent capabilities to the systems we already have?

Embedding an agent into an existing system means plugging an LLM-driven reasoning loop into infrastructure that predates agent architecture. The agent needs to receive triggers from the existing system, access its data, take actions through its APIs, and return results in formats the system understands. The agent is a new component in an old architecture, and the integration patterns matter more than the agent's internal design.

Integration Surface Areas #

An agent that lives inside an existing system touches it at four points:

┌──────────────────────────────────────────────────────────┐
│                    Existing System                       │
│                                                          │
│  ┌────────────┐   ┌────────────┐   ┌────────────────┐    │
│  │  Triggers  │   │    Data    │   │    Actions     │    │
│  │            │   │            │   │                │    │
│  │ HTTP hooks │   │ Databases  │   │ APIs           │    │
│  │ Queue msgs │   │ File stores│   │ DB writes      │    │
│  │ Cron jobs  │   │ Caches     │   │ Notifications  │    │
│  │ DB changes │   │ Logs       │   │ Deployments    │    │
│  └──────┬─────┘   └──────┬─────┘   └───────┬────────┘    │
│         │                │                 │             │
│         ▼                ▼                 ▼             │
│  ┌──────────────────────────────────────────────────┐    │
│  │                  Agent Layer                     │    │
│  │   Trigger ──► Reason ──► Act ──► Respond         │    │
│  └──────────────────────────────────────────────────┘    │
│         │                                                │
│         ▼                                                │
│  ┌────────────┐                                          │
│  │  Results   │  Writes, messages, status updates,       │
│  │            │  transformed data, human notifications   │
│  └────────────┘                                          │
└──────────────────────────────────────────────────────────┘

Triggers — what causes the agent to run. An HTTP request, a message on a queue, a cron schedule, a database change, a webhook from a third-party service.

Data — what the agent reads. Database tables, file stores, caches, configuration, logs, external APIs. The agent needs read access to whatever context it requires for reasoning.

Actions — what the agent does. Write to a database, call an API, send a notification, modify a file, trigger a deployment. These tools operate through the existing system's interfaces.

Results — what the agent produces. A response message, a status update, a modified record, a generated artifact. The result must fit the existing system's expected formats and protocols.

Agents as Microservices #

The most common integration pattern is deploying the agent as a service behind an HTTP API. The existing system calls the agent like any other microservice — send a request, get a response.

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field


app = FastAPI()


class AgentRequest(BaseModel):
    task: str
    context: dict = Field(default_factory=dict)
    timeout_seconds: float = Field(default=60.0, gt=0)


class AgentResponse(BaseModel):
    result: str
    success: bool
    steps_taken: int
    tokens_used: int


@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
    output = await agent.run(AgentInput(
        task=request.task,
        context=request.context,
        timeout_seconds=request.timeout_seconds,
    ))
    return AgentResponse(
        result=output.result,
        success=output.success,
        steps_taken=output.steps_taken,
        tokens_used=output.tokens_used,
    )

The calling system sees a stable contract: it sends a JSON request and gets a JSON response. A standard API boundary encapsulates the agent's nondeterminism, tool calls, and multi-step reasoning.

Synchronous vs. Asynchronous #

Short agent tasks (classification, summarization, simple tool calls) fit the synchronous request-response model. The caller waits, the agent runs, and the response comes back in seconds.

Longer tasks need an asynchronous pattern. The caller submits a job, gets back a job ID, and polls or receives a callback when the work is done.

import uuid
import httpx
from fastapi import HTTPException


jobs: dict[str, dict] = {}


class JobSubmission(BaseModel):
    task: str
    context: dict = Field(default_factory=dict)
    callback_url: str | None = None


class JobStatus(BaseModel):
    job_id: str
    status: str  # "pending", "running", "completed", "failed"
    result: str | None = None


@app.post("/agent/submit")
async def submit_job(
    submission: JobSubmission,
    background_tasks: BackgroundTasks,
):
    job_id = str(uuid.uuid4())
    jobs[job_id] = {"status": "pending", "result": None}

    background_tasks.add_task(
        execute_agent_job, job_id, submission
    )

    return {"job_id": job_id, "status": "pending"}


@app.get("/agent/status/{job_id}", response_model=JobStatus)
async def get_job_status(job_id: str):
    job = jobs.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return JobStatus(job_id=job_id, **job)


async def execute_agent_job(job_id: str, submission: JobSubmission):
    jobs[job_id]["status"] = "running"
    try:
        output = await agent.run(AgentInput(
            task=submission.task,
            context=submission.context,
        ))
        jobs[job_id] = {"status": "completed", "result": output.result}

        if submission.callback_url:
            async with httpx.AsyncClient() as client:
                await client.post(submission.callback_url, json={
                    "job_id": job_id,
                    "result": output.result,
                })
    except Exception as e:
        jobs[job_id] = {"status": "failed", "result": str(e)}

The async pattern with callbacks is how agents integrate with pipeline orchestrators, CI/CD systems, and event-driven architectures. The calling system fires and forgets; the agent calls back when done.

Event-Driven Triggers #

Many existing systems already emit events — messages on a queue, webhook notifications, change streams. Agents can subscribe to these events and activate when relevant things happen.

Message Queue Integration #

import json
import aio_pika


async def start_queue_consumer(
    queue_name: str,
    agent: "Agent",
    connection_url: str,
):
    """
    Consume messages from a queue and route each
    to the agent for processing.
    """
    connection = await aio_pika.connect_robust(connection_url)
    async with connection:
        channel = await connection.channel()
        queue = await channel.declare_queue(queue_name, durable=True)

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    payload = json.loads(message.body)
                    task = payload.get("task", "")
                    context = payload.get("context", {})

                    output = await agent.run(AgentInput(
                        task=task,
                        context=context,
                    ))

                    # Publish result to a response queue
                    if payload.get("reply_to"):
                        await channel.default_exchange.publish(
                            aio_pika.Message(
                                body=json.dumps({
                                    "request_id": payload.get("request_id"),
                                    "result": output.result,
                                    "success": output.success,
                                }).encode(),
                            ),
                            routing_key=payload["reply_to"],
                        )

The agent becomes a queue consumer. Upstream systems publish messages through familiar queue semantics, the agent processes them, and downstream systems consume results from the response queue. The agent is just another worker in the existing message-driven architecture.

When to use: high-throughput scenarios where many events need agent processing, systems that already use queues for decoupling, and cases where agent processing time is unpredictable and you need backpressure.

Database Change Triggers #

Some of the most natural agent triggers are database changes. A new row appears in the support_tickets table — an agent should classify and route it. An order's status changes to flagged — an agent should review it for fraud. A document is uploaded to a table — an agent should extract metadata.

import asyncio
import asyncpg


async def watch_for_changes(
    dsn: str,
    channel: str,
    agent: "Agent",
):
    """
    Listen for PostgreSQL NOTIFY events and trigger
    agent processing for each change.
    """
    allowed_tables = {"support_tickets"}
    notifications: asyncio.Queue[str] = asyncio.Queue()

    def on_notification(connection, pid, notified_channel, payload):
        notifications.put_nowait(payload)

    conn = await asyncpg.connect(dsn)
    await conn.add_listener(channel, on_notification)

    try:
        while True:
            payload = json.loads(await notifications.get())

            table = payload.get("table")
            operation = payload.get("operation")  # INSERT, UPDATE
            record_id = payload.get("id")
            if table not in allowed_tables or record_id is None:
                continue

            # The allowlist makes this identifier interpolation safe.
            record = await conn.fetchrow(
                f"SELECT * FROM {table} WHERE id = $1", record_id
            )
            if record is None:
                continue

            task = build_task_from_change(table, operation, dict(record))
            output = await agent.run(AgentInput(task=task))

            # Write the agent's result back.
            await conn.execute(
                f"UPDATE {table} SET agent_analysis = $1, "
                f"agent_processed_at = NOW() WHERE id = $2",
                output.result, record_id,
            )
    finally:
        await conn.remove_listener(channel, on_notification)
        await conn.close()

On the database side, a trigger function emits notifications:

CREATE OR REPLACE FUNCTION notify_agent_trigger()
RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(
        'agent_changes',
        json_build_object(
            'table', TG_TABLE_NAME,
            'operation', TG_OP,
            'id', NEW.id
        )::text
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER support_ticket_agent
    AFTER INSERT OR UPDATE ON support_tickets
    FOR EACH ROW EXECUTE FUNCTION notify_agent_trigger();

The agent reacts to database changes through a trigger. The existing application inserts a support ticket through its normal workflow; the trigger fires; the agent processes the ticket and writes its analysis back to the same row. From the application's perspective, tickets simply get classified automatically.

Watch out for: feedback loops. If the agent writes back to the same table that triggered it, the trigger fires again. Guard against this with a condition (WHEN NEW.agent_processed_at IS NULL) or a separate "agent results" table.

Agents in CI/CD Pipelines #

CI/CD pipelines are a natural fit for agents because they already have the infrastructure for triggering work, passing artifacts between steps, and reporting results.

┌──────────┐     ┌──────────┐    ┌──────────────┐     ┌──────────┐
│  Commit  │───▶│  Build   │───▶│  Agent Step  │───▶│  Deploy  │
│          │     │  & Test  │    │              │     │          │
│          │     │          │    │ Code review  │     │          │
│          │     │          │    │ or security  │     │          │
│          │     │          │    │ analysis     │     │          │
└──────────┘     └──────────┘    └──────────────┘     └──────────┘

A CI/CD agent step receives pipeline context (changed files, test results, build logs) and produces a structured result (approval, findings, comments).

import json


def ci_agent_step(
    pipeline_context: dict,
    agent: "Agent",
) -> dict:
    """
    Run an agent as a CI/CD pipeline step.
    Receives pipeline context, returns structured findings.
    """
    changed_files = pipeline_context.get("changed_files", [])
    test_results = pipeline_context.get("test_results", "")
    build_log_tail = pipeline_context.get("build_log_tail", "")

    task = (
        f"Review the following code changes for security issues "
        f"and potential bugs.\n\n"
        f"Changed files:\n{json.dumps(changed_files, indent=2)}\n\n"
        f"Test results:\n{test_results}\n\n"
        f"Build log (last 50 lines):\n{build_log_tail}\n\n"
        "Return JSON with three array fields: blockers, warnings, and comments."
    )

    output = agent.run_sync(AgentInput(
        task=task,
        max_steps=10,
        timeout_seconds=120,
    ))

    # Parse structured output for the pipeline
    findings = parse_findings(output.result)

    return {
        "status": "pass" if not findings["blockers"] else "fail",
        "blockers": findings["blockers"],
        "warnings": findings["warnings"],
        "comments": findings["comments"],
    }


def parse_findings(agent_output: str) -> dict:
    """
    Extract structured findings from agent output.
    The agent is prompted to output a specific format.
    """
    try:
        parsed = json.loads(agent_output)
    except json.JSONDecodeError:
        return {
            "blockers": [],
            "warnings": [agent_output],
            "comments": [],
        }
    if not isinstance(parsed, dict):
        return {
            "blockers": [],
            "warnings": ["Agent returned JSON with the wrong top-level type"],
            "comments": [],
        }

    def string_list(key: str) -> list[str]:
        value = parsed.get(key, [])
        if not isinstance(value, list):
            return [f"Agent returned a non-list value for '{key}'"]
        return [str(item) for item in value]

    return {
        "blockers": string_list("blockers"),
        "warnings": string_list("warnings"),
        "comments": string_list("comments"),
    }

The agent step returns a pass/fail status that the pipeline understands. If the agent finds blocking issues, the pipeline stops — just like a failing test would. Warnings and comments can be posted to the pull request or recorded as build artifacts.

Pipeline Integration Patterns #

Different CI/CD systems have different extension mechanisms:

Step-based pipelines (Jenkins, GitHub Actions, GitLab CI) — the agent runs as a custom step in the pipeline definition. It receives inputs from previous steps and produces outputs for subsequent ones.

Plugin/extension model — wrap the agent in a pipeline plugin that handles input parsing, output formatting, and status reporting according to the platform's conventions.

Webhook-triggered — the CI system fires a webhook when a stage completes. The agent service receives the webhook, processes the artifacts, and posts results back via the CI system's API.

# Example: agent step in a pipeline definition
steps:
  - name: build
    run: make build

  - name: test
    run: make test

  - name: agent-review
    uses: ./agent-step
    with:
      task: "Review changes for security and correctness"
      timeout: 120
      fail-on-blockers: true

  - name: deploy
    run: make deploy
    if: steps.agent-review.outputs.status == 'pass'

Wrapping Existing APIs as Agent Tools #

The fastest way to give an agent access to an existing system is to wrap the system's APIs as tools. The agent's tool definitions become a thin layer over existing endpoints.

import httpx
from urllib.parse import quote


def wrap_api_as_tool(
    name: str,
    description: str,
    base_url: str,
    method: str,
    path: str,
    parameters: dict,
    auth_header: str | None = None,
) -> "ToolDefinition":
    """
    Wrap an existing HTTP API endpoint as an agent tool.
    """

    async def call_api(**kwargs) -> str:
        url = f"{base_url}{path}"
        request_data = dict(kwargs)

        # Substitute path parameters
        for key, value in list(request_data.items()):
            if f"{{{key}}}" in url:
                url = url.replace(f"{{{key}}}", quote(str(value), safe=""))
                request_data.pop(key)

        if "{" in url or "}" in url:
            return "API error: missing a required path parameter"

        headers = {}
        if auth_header:
            headers["Authorization"] = auth_header

        async with httpx.AsyncClient(timeout=10.0) as client:
            if method.upper() == "GET":
                response = await client.get(url, params=request_data, headers=headers)
            elif method.upper() == "POST":
                response = await client.post(url, json=request_data, headers=headers)
            else:
                response = await client.request(
                    method.upper(), url, json=request_data, headers=headers,
                )

        if response.status_code >= 400:
            return f"API error {response.status_code}: {response.text[:500]}"

        return response.text[:4000]

    return ToolDefinition(
        name=name,
        description=description,
        parameters=parameters,
        fn=call_api,
    )


# Wrap an existing order service
get_order_tool = wrap_api_as_tool(
    name="get_order",
    description="Look up an order by ID. Returns order details including status and items.",
    base_url="http://order-service:8080",
    method="GET",
    path="/api/orders/{order_id}",
    parameters={
        "type": "object",
        "properties": {
            "order_id": {"type": "string", "description": "The order ID to look up"}
        },
        "required": ["order_id"],
    },
)

The key principle: the agent joins the system as a new API client. The order service, the inventory system, and the notification service all continue to work exactly as they did while the agent consumes their existing interfaces.

Tool Boundaries and Safety #

When wrapping existing APIs as tools, the security boundary matters more than usual because the agent is now connected to production systems.

Read vs. write separation. Start by giving the agent read-only tools. Let it query orders, look up customers, and search logs. Add write tools (cancel order, issue refund, update record) only after you have validated the agent's behavior on read-only tasks with human-in-the-loop approval for destructive operations.

Rate limiting. The existing API may be sensitive to the burst patterns an agent produces. A ReAct loop might call the same endpoint ten times in ten seconds. Tool-wrapper rate limits complement the API's own limits.

Credential isolation. The agent should authenticate with its own service account. This gives you a clear audit trail and lets you apply specific permission boundaries to the agent's access level.

Incremental Adoption #

The mistake teams make is trying to build a comprehensive agent system from day one. The pattern that works is incremental: start with one narrow task, prove it works, then expand.

Phase 1: Single task, read-only
  └─ Agent classifies support tickets
  └─ Writes classification to a new column
  └─ Humans review and correct

Phase 2: Add actions
  └─ Agent routes tickets to the right team
  └─ Auto-responds to known simple questions
  └─ Escalates uncertain cases to humans

Phase 3: Expand scope
  └─ Agent handles order inquiries (reads order data)
  └─ Agent drafts responses for human approval
  └─ Agent accesses knowledge base for answers

Phase 4: Autonomous operation
  └─ Agent resolves straightforward tickets end-to-end
  └─ Humans handle edge cases and review agent decisions
  └─ Agent learns from corrections (memory, few-shot examples)

Each phase has a clear scope, a clear success metric, and a rollback path. Phase 1 is safe — the agent reads data and writes to a new field isolated from existing consumers. Classification errors remain contained within that field. Phase 2 introduces actions with human oversight. Phase 3 expands data access. Phase 4 grants autonomy only for proven task types.

This incremental approach also surfaces previously hidden integration problems: authentication quirks, API rate limits, data format mismatches, and edge cases in the business logic. A single read-only task provides a safer place to discover them than a fully autonomous agent connected to your production database.

Sidecar and Middleware Patterns #

For systems where modifying the core application is expensive or risky, agents can operate as sidecars — separate processes that observe and augment the primary system while preserving its core flow.

┌────────────────────────────────────────────────┐
│           Primary Application (unchanged)      │
│                                                │
│  Request ──► Business Logic ──► Response       │
│      │                              │          │
│      │         Mirrored             │          │
│      ▼                              ▼          │
│  ┌──────────────────────────────────────────┐  │
│  │          Agent Sidecar                   │  │
│  │                                          │  │
│  │  Observes requests and responses.        │  │
│  │  Enriches, classifies, or flags.         │  │
│  │  Writes to a separate data store.        │  │
│  │  Never modifies the primary flow.        │  │
│  └──────────────────────────────────────────┘  │
└────────────────────────────────────────────────┘

The sidecar reads the same inputs and outputs as the primary application while processing them independently. It might classify incoming requests, flag anomalies, generate summaries of activity, or build a knowledge base from observed interactions. The primary application's control flow remains unchanged.

When to use: legacy systems with a fixed core, high-risk environments where the primary flow requires strong stability, and exploratory phases where you want to evaluate agent behavior before integrating it into the main path.

A middleware variant intercepts requests in the flow — reading and optionally enriching them before they reach the primary application. This is more powerful but riskier: the agent is now in the critical path, and its latency and reliability affect the primary system.

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response


class AgentEnrichmentMiddleware(BaseHTTPMiddleware):
    """
    Middleware that enriches incoming requests with
    agent-generated metadata before they reach the app.
    """

    def __init__(self, app, agent, enrichment_paths: list[str]):
        super().__init__(app)
        self.agent = agent
        self.enrichment_paths = enrichment_paths

    async def dispatch(self, request: Request, call_next) -> Response:
        if request.url.path not in self.enrichment_paths:
            return await call_next(request)

        body = await request.body()
        try:
            enrichment = await self.agent.run(AgentInput(
                task=f"Classify this request: {body.decode()[:1000]}",
                max_steps=3,
                timeout_seconds=5,
            ))
            request.state.agent_classification = enrichment.result
        except Exception:
            request.state.agent_classification = None

        return await call_next(request)

The middleware adds optional agent-generated metadata to the request. The primary application can use it or ignore it. If the agent fails or times out, the request proceeds along the primary flow in its original form.

Conclusion #

Embedding agents in existing systems is an integration problem more than an AI problem. The agent's internal design — its prompts, tools, and orchestration loop — matters, but the integration patterns determine whether it works in practice:

• Agents as microservices give you the cleanest boundary. The existing system calls the agent through HTTP, which encapsulates the agent's internals. Use synchronous calls for fast tasks and async job submission for slow ones.
• Event-driven triggers let agents react to queue messages, database changes, and webhooks while preserving the code that produces those events.
• Database change triggers are especially natural. The existing application writes data through its normal path; the agent processes changes and writes results back. Guard against feedback loops.
• CI/CD pipeline steps give agents structured context (changed files, test results, build logs) and expect structured output (pass/fail, findings, comments).
• Existing APIs become tools with thin wrappers. Start read-only, add writes incrementally, and isolate agent credentials from human credentials.
• Incremental adoption is the only approach that works reliably. Start with one read-only task, prove the value, then expand scope and autonomy phase by phase.
• Sidecar patterns add agent capabilities around systems with a fixed core. The agent observes and augments the surrounding flow.

The recurring principle: the agent adapts to the system. Existing systems have proven reliability, known failure modes, and teams that understand them. The agent earns its place by working within those constraints and fitting established interfaces.

Continue exploring

• Agent Frameworks - Build vs. Buy Trade-offs between agent frameworks and custom orchestration. Shared tags: ai-agents · Same collection · Same subsection
• Streaming, Speculative Execution & Latency Engineering Reducing real and perceived latency in multi-step agent systems. Shared tags: ai-agents · Same collection · Same subsection
• Error Handling & Recovery Failures classification, retry safely, fall back, checkpoint, and recover. Shared tags: ai-agents · Same collection · Same subsection