Metadata-Version: 2.4
Name: pydantic-ai-triggers
Version: 0.1.0
Summary: Event triggers for Pydantic AI agents
Project-URL: Repository, https://github.com/joeychilson/pydantic-ai-triggers
Project-URL: Issues, https://github.com/joeychilson/pydantic-ai-triggers/issues
Author-email: Joey Chilson <joe.chilson@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,cron,pydantic,pydantic-ai,triggers,webhook
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: croniter>=6.2.2
Requires-Dist: pydantic-ai>=1.0
Requires-Dist: starlette>=1.3.1
Requires-Dist: uvicorn>=0.49.0
Description-Content-Type: text/markdown

# pydantic-ai-triggers

Event triggers for [Pydantic AI](https://ai.pydantic.dev) agents. Wire an agent to
an event source with a decorator-based API that feels native to the library: import
`Agent` from here instead of `pydantic_ai`, register handlers with `@agent.on(...)`,
and run them.

```python
import asyncio

from pydantic_ai_triggers import Agent, Cron, TriggerContext

agent = Agent("anthropic:claude-opus-4-8")


@agent.on(Cron("0 9 * * *", timezone="America/Chicago"))
async def daily_digest(ctx: TriggerContext[None]) -> str:
    # The return value is the prompt the agent runs with.
    return "Summarize yesterday's activity."


if __name__ == "__main__":
    asyncio.run(agent.run_triggers())
```

A handler receives a `TriggerContext` (the agent's `deps` and the event that fired)
and returns the prompt to run the agent with. `run_triggers()` runs every registered
trigger concurrently until cancelled, logging each agent run on the
`pydantic_ai_triggers` logger.

## Triggers

- **`Cron`** — a polling schedule (`Cron("0 9 * * *", timezone=...)`).
- **`Webhook`** — an inbound HTTP request. The handler's `ctx.event` is a typed
  `WebhookEvent` (`ctx.event.json`, `.headers`, `.body`, …):

  ```python
  from pydantic_ai_triggers import Agent, Webhook, TriggerContext, WebhookEvent

  agent = Agent("anthropic:claude-opus-4-8")


  @agent.on(Webhook("/webhooks/github"))
  async def triage(ctx: TriggerContext[None, WebhookEvent]) -> str:
      issue = ctx.event.json["issue"]
      return f"Triage this issue:\n\n{issue['title']}"
  ```

  `run_triggers()` serves webhooks on a built-in HTTP server alongside any polling
  triggers; or mount them yourself with `app = agent.to_trigger_app()` (an ASGI app)
  for serverless or an existing framework. The agent runs within the request, so
  keep handlers quick or offload slow work.

> **Status:** early development. API is not yet stable. `Cron` and `Webhook` triggers
> exist today; a typed GitHub trigger is planned.

> Unofficial — not affiliated with or endorsed by Pydantic.

## Examples

See [`examples/`](examples/). They live outside the package and add no
dependencies — [`openrouter_cron.py`](examples/openrouter_cron.py) runs a
cron-triggered health check against [OpenRouter](https://openrouter.ai) models,
showing typed deps, skipping a firing, switching models per firing, and structured
output, using only what `pydantic-ai` already provides:

```bash
export OPENROUTER_API_KEY="sk-or-..."
uv run python examples/openrouter_cron.py
```

[`webhook.py`](examples/webhook.py) serves a webhook-triggered agent and runs
offline with a stub model (no API key needed):

```bash
uv run python examples/webhook.py
# then: curl -X POST localhost:8000/github -d '{"action":"opened","issue":{"title":"..."}}'
```

## Development

This project uses [uv](https://docs.astral.sh/uv/), [ruff](https://docs.astral.sh/ruff/),
and [ty](https://github.com/astral-sh/ty).

```bash
uv sync                  # create the venv and install everything
uv run ruff check .      # lint
uv run ruff format .     # format
uv run ty check          # type-check
uv run pytest            # tests
```
