Middleware
Middleware runs custom logic before and after agent execution to attach documents, enrich context, validate inputs, and transform responses.
On this page
What is Middleware?
Hooks that run before and after the agent processes a request
Middleware are Python functions that intercept requests before they reach the agent and responses before they're returned. Use them to attach documents, enrich context, validate inputs, log interactions, or transform responses. Middleware and tool hooks can call predefined tools from connic.tools, including the browser tools. A browser opened in before() uses the session's saved browser profile when the session key can already be resolved.
Create a file in middleware/ with the same name as the agent. For example, middleware/assistant.py applies to the agent whose YAML has name: assistant, even if that YAML lives in a nested folder under agents/. No YAML configuration is required.
Execution Flow
Basic Middleware
"""Middleware for the assistant agent."""
import json
from datetime import datetime
from typing import Any, Dict
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""
Called before the agent runs.
Args:
content: Dict with "role" and "parts" keys
context: Shared run context dict (see Context docs)
Returns:
Modified content to pass to the agent
"""
print(f"Before middleware running for run {context.get('run_id')}")
# Add current timestamp so the agent knows today's date
now = datetime.now()
time_context = f"[Current time: {now.strftime('%A, %B %d, %Y')}]"
content["parts"].insert(0, {"text": time_context})
return content
async def after(response: str, context: Dict[str, Any]) -> str:
"""
Called after the agent completes.
Args:
response: The agent's response text
context: Shared run context dict (see Context docs)
Returns:
Modified response (or original)
"""
print(f"After middleware — response length: {len(response)}")
# Wrap response in structured JSON with metadata
return json.dumps({
"response": response,
"run_id": context.get("run_id"),
"duration_ms": context.get("duration_ms"),
})Every hook is optional. Define any combination of before(), after(), and cleanup().before() must return content; the value returned by after() becomes the response.
The Content Dict
The before() hook receives a content dict that represents the user's message. It contains a role (always 'user') and a list of parts. Each part is a dict with either text or binary data.
# The content dict structure
content = {
"role": "user", # Always 'user' — before() receives the inbound message
"parts": [ # List of part dicts
{"text": "Hello, analyze this document"},
{"data": b"...", "mime_type": "application/pdf"},
]
}
# Accessing content parts
for part in content["parts"]:
if "text" in part:
# Text content
text = part["text"]
elif "data" in part:
# Binary file content
mime_type = part["mime_type"]
data = part["data"]The Original Payload
The raw connector payload is also available as context["payload"]. Use it for request metadata, authentication fields, webhook JSON bodies, form fields, or GET query parameters. Use content to change what the agent sees.
# The original connector payload is available in context
payload = context.get("payload", {})
# For a JSON webhook body:
# {"auth_token": "abc", "question": "Hello"}
# payload["auth_token"] == "abc"
# payload["question"] == "Hello"
# For a GET webhook:
# <webhook-url>?user_id=123&query=hello
# payload["user_id"] == "123"
# payload["query"] == "hello"Creating Parts
Create parts as simple dicts with text for text content or data + mime_type for files.
# Creating parts as dicts
# Text part
text_part = {"text": "Analyze this document"}
# File from bytes (PDFs, images, audio, video)
with open("document.pdf", "rb") as f:
pdf_part = {"data": f.read(), "mime_type": "application/pdf"}
# Image from bytes
with open("image.png", "rb") as f:
image_part = {"data": f.read(), "mime_type": "image/png"}
# Adding parts to content
content["parts"].append(pdf_part)
content["parts"].insert(0, text_part)Common Use Cases
Attaching Documents Dynamically
Add PDFs, images, or other files to the agent's context based on the user's request.
"""Attach documents dynamically based on request context."""
from typing import Any, Dict
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Attach relevant documents to the agent context."""
# Get the user's text message
user_text = ""
for part in content["parts"]:
if "text" in part:
user_text = part["text"]
break
# Attach a retrieval document based on topic
if "pricing" in user_text.lower():
with open("docs/pricing.pdf", "rb") as f:
content["parts"].append({
"data": f.read(),
"mime_type": "application/pdf"
})
elif "technical" in user_text.lower():
with open("docs/technical-specs.pdf", "rb") as f:
content["parts"].append({
"data": f.read(),
"mime_type": "application/pdf"
})
return contentAdding Context & Customer Data
Prepend system context or fetch data from external APIs to enrich the request.
"""Add system context and customer data to requests."""
from typing import Any, Dict
import httpx
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Prepend system context to the user message."""
# Read request data from the original connector payload.
payload = context.get("payload", {})
customer_id = payload.get("customer_id")
# Fetch customer data from your API
customer_info = ""
if customer_id:
print(f"Fetching customer data for {customer_id}")
async with httpx.AsyncClient() as client:
resp = await client.get(f"https://api.example.com/customers/{customer_id}")
if resp.status_code == 200:
data = resp.json()
customer_info = f"""
Customer: {data['name']}
Plan: {data['plan']}
Account Status: {data['status']}
"""
print(f"Loaded customer: {data['name']} ({data['plan']})")
# Prepend context as text part
content["parts"].insert(0, {
"text": f"[CUSTOMER CONTEXT]\n{customer_info}\n[END CONTEXT]"
})
return contentReplacing the Message Text
File uploads arrive as separate parts next to the text part. To change what the agent reads, replace only the text part. Setting content["parts"] to a text-only list drops every attachment.
"""Show the agent only the chat message and keep uploaded files."""
from typing import Any, Dict
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
# Webhook body: {"user_id": "u123", "message": "What is in this file?"} plus uploads
payload = context.get("payload", {})
context["user_id"] = payload.get("user_id")
# Replace only the text part; file parts stay in the list
files = [part for part in content["parts"] if "text" not in part]
content["parts"] = [{"text": payload.get("message", "")}, *files]
return contentGraceful Stop with StopProcessing
"""Authentication middleware - stops processing if invalid."""
from typing import Any, Dict
from connic import StopProcessing
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Validate request before agent runs."""
# Read the original connector payload. For webhooks, this is the JSON body,
# form fields, or GET query parameters.
payload = context.get("payload", {})
auth_token = payload.get("auth_token")
if not auth_token:
raise StopProcessing("Authentication required")
# Store verified request data in context for prompts and tools.
context["user_id"] = verify_auth_token(auth_token)
return contentStopProcessing vs Exceptions: StopProcessing returns a response gracefully (run marked as completed). Regular exceptions mark the run as failed. Because the run completes, automatic outbound connectors still deliver it by default, with the StopProcessing response as the output. Pass publish_outbound=False to skip the automatic outbound connector for that graceful stop. This flag does not cancel an agent-tool or middleware outbound connector call. The same mechanism works for tools in tools/. See Write Tools
Analytics & Logging
"""Send analytics to external service after agent completes."""
import httpx
from typing import Any, Dict
async def after(response: str, context: Dict[str, Any]) -> str:
"""Send analytics after agent completes."""
try:
async with httpx.AsyncClient() as client:
await client.post(
"https://analytics.internal/events",
json={
"event": "agent_run_completed",
"run_id": context.get("run_id"),
"agent": context.get("agent_name"),
"duration_ms": context.get("duration_ms"),
"tokens": context.get("token_usage", {}),
}
)
except Exception:
pass # Don't fail the request if analytics fails
return responseAny print or stdlib logging call emitted from before(), after(), or cleanup() is captured per run and shown on the project Logs tab and in the run detail view, tagged with source middleware.before, middleware.after, or middleware.cleanup.
Middleware Outbound Connectors
Set an outbound connector's mode to Middleware, give it a call name, then call send_connector(action_name, payload) from before() or after(). Use this when code, not the model, should decide whether to send.
"""Send an urgent result through a configured middleware outbound connector."""
from typing import Any, Dict
from connic.tools import send_connector
async def after(response: str, context: Dict[str, Any]) -> str:
payload = context.get("payload", {})
if payload.get("priority") == "urgent":
await send_connector(
"send_to_ops_slack",
{"text": response},
)
return responseThe payload must match that outbound connector's documented payload schema. connic resolves the connector, destination, credentials, formatting, and retries server-side; middleware receives none of those stored values. Pass idempotency_key=... when the code may retry the same call. See outbound connector modes.
Response Transformation
"""Transform response format for specific use cases."""
import json
from typing import Any, Dict
def after(response: str, context: Dict[str, Any]) -> str:
"""Wrap response in a standard API format."""
return json.dumps({
"success": True,
"data": response,
"metadata": {
"run_id": context.get("run_id"),
"processed_at": context.get("timestamp"),
}
})Cleanup After Every Run
after() only runs when the agent finishes. To release something that before() set up, such as a lock or a claimed record, define cleanup(run, context). It runs once the run has ended, whether it completed, failed, timed out, or was cancelled.
"""Claim an order while the agent works on it and release it however the run ends."""
from typing import Any, Dict
import httpx
from connic import StopProcessing
ORDERS_API = "https://api.example.com/orders"
async def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
order_id = context["payload"]["order_id"]
async with httpx.AsyncClient() as client:
resp = await client.post(f"{ORDERS_API}/{order_id}/claim")
if resp.status_code == 409:
raise StopProcessing("Order is already being processed")
context["claimed_order_id"] = order_id
return content
async def cleanup(run: Dict[str, Any], context: Dict[str, Any]) -> None:
order_id = context.get("claimed_order_id")
if not order_id:
return # before() stopped or failed before it claimed the order
async with httpx.AsyncClient() as client:
await client.post(
f"{ORDERS_API}/{order_id}/release",
json={"status": run["status"], "error": run["error"]},
)run["status"] is completed, failed, or cancelled. A timeout counts as failed, and a StopProcessing stop as completed. run["error"] holds the error message the run shows, or None, and run["output"] the final response.
- Once per run, after
after()and output guardrails. Sync connectors return the response aftercleanup()finishes, so keep it short. It has 30 seconds. - Also when
before()stopped or failed. Set acontextkey oncebefore()holds the resource and check it incleanup(), as above. - Not while the run waits for an approval or a sequential step. It runs when the resumed run ends.
- Errors and timeouts in
cleanup()are logged on the run and never change its status.
A run cancelled before it starts never calls cleanup(), and neither does a run that fails or is cancelled while it waits for an approval. If the agent process stops unexpectedly, cleanup() can't run either. For locks, also store when they were taken and treat old ones as expired.
The context dict is shared and mutable. Values set in before() are available in prompts via {var} syntax, in tools, and in the after() hook. See the Context documentation for full details.
Project Structure
Sync and Async Support
"""Sync middleware also works."""
from typing import Any, Dict
def before(content: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Sync functions are automatically handled."""
# Add a simple prefix to the first text part
for part in content["parts"]:
if "text" in part:
part["text"] = "[Processed] " + part["text"]
break
return content
def after(response: str, context: Dict[str, Any]) -> str:
"""Both sync and async are supported."""
return response.strip()Use async functions for I/O operations (API calls, database queries, file reads). Sync functions are fine for simple transformations.
If a middleware function raises an exception, the entire request fails. The run is marked as failed, and for before() the agent never executes. Use middleware as a gatekeeper for validation and authentication. cleanup() is the exception: its errors are only logged.
By default, before() runs once before execution. For tool and sequential agents, set retry_options.rerun_middleware: true to re-run it on operation retries.
LLM model-request retries, fallback transitions, and tool-failure reflection stay inside the current agent run. They never re-run middleware, so rerun_middleware is ignored for type: llm.
When enabled, each retry rebuilds the content from the original payload and calls before() again, allowing middleware to refresh dynamic context (e.g., fetching external state after partial tool calls). The context dict accumulates state across retries. Use context.get("retry_attempt", 0) to check which attempt is running. The after() middleware runs once after all retries complete unless before() raises StopProcessing.cleanup() also runs once, after the last attempt.