pydantic-ai-middleware (Abandoned: Renamed to pydantic-ai-shields)

0.2.4 · abandoned · verified Fri Apr 17

`pydantic-ai-middleware` was a library providing a simple middleware layer for Pydantic-AI agents, allowing for `before_run`/`after_run` hooks and other lifecycle management without imposing strict guardrails. The last version of this package was `0.2.4`. As of version `0.3.0`, the package was renamed to `pydantic-ai-shields`, and its core middleware functionality was integrated directly into `pydantic-ai` v1.71+.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to wrap a `pydantic-ai` agent with custom middleware using `MiddlewareAgent` and `AgentMiddleware` from `pydantic-ai-middleware` v0.2.4. It defines a simple middleware to log before and after agent execution.

import os
from pydantic_ai import PydanticAI
from pydantic_ai.agents import Agent
from pydantic_ai_middleware.agent import MiddlewareAgent
from pydantic_ai_middleware.middleware import AgentMiddleware
import asyncio

class MyCustomMiddleware(AgentMiddleware):
    async def before_run(self, agent: Agent, input_data: str, **kwargs):
        print(f"Middleware: Before run with input: {input_data}")
        return input_data

    async def after_run(self, agent: Agent, output_data: str, **kwargs):
        print(f"Middleware: After run with output: {output_data}")
        return output_data

class MyAgent(Agent):
    def run(self, topic: str) -> str:
        """Generate a short story about the given topic."""
        return f"A story about {topic}..."

ai = PydanticAI(
    api_key=os.environ.get("OPENAI_API_KEY", "sk-DUMMY"), # Use a dummy key if not set
    model="gpt-3.5-turbo",
)
my_agent = MyAgent(ai=ai)

middleware_agent = MiddlewareAgent(
    wrapped_agent=my_agent,
    middleware=[MyCustomMiddleware()]
)

async def main():
    print("\n--- Running Middleware Agent ---")
    result = await middleware_agent.run("a brave knight")
    print(f"Agent final result: {result}")
    print("--------------------------------")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →