Integrations¶

from fastagentic import App
from fastagentic.integrations.langfuse import LangfuseHook
from fastagentic.integrations.lakera import LakeraHook
from fastagentic.integrations.mem0 import Mem0Provider

app = App(
    title="My Agent",
    version="1.0.0",

    # Global hooks (apply to all endpoints)
    hooks=[
        LangfuseHook(
            public_key="pk-...",
            secret_key="sk-...",
        ),
        LakeraHook(
            api_key="...",
            on_failure="warn",  # fail-open
        ),
    ],

    # Memory provider
    memory=Mem0Provider(api_key="..."),
)

from fastagentic import agent_endpoint
from fastagentic.integrations.guardrails import GuardrailsAIHook
from fastagentic.integrations.braintrust import BraintrustHook

@agent_endpoint(
    path="/triage",
    runnable=...,
    post_hooks=[
        GuardrailsAIHook(rail="triage_output.rail"),
    ],
    eval_hooks=[
        BraintrustHook(project="support-triage"),
    ],
)
async def triage(ticket: TicketIn) -> TicketOut:
    ...

from fastagentic.integrations.langfuse import LangfuseHook

# Automatically reads from environment
app = App(hooks=[LangfuseHook()])

from fastagentic import App
from fastagentic.config import load_config

config = load_config("config/settings.yaml")
app = App.from_config(config)

# Fail-closed: Block if Lakera fails
LakeraHook(api_key="...", on_failure="reject")

# Fail-open: Log warning, continue if Lakera fails
LakeraHook(api_key="...", on_failure="warn")

from fastagentic.hooks import BaseHook, HookContext

class MyCustomHook(BaseHook):
    hooks = ["on_llm_start", "on_llm_end"]

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.client = MyClient(api_key)

    async def on_llm_start(self, ctx: HookContext):
        # Pre-LLM logic
        pass

    async def on_llm_end(self, ctx: HookContext):
        # Post-LLM logic
        await self.client.track(
            model=ctx.model,
            tokens=ctx.usage.total_tokens,
        )

from fastagentic.memory import MemoryProvider

class MyMemoryProvider(MemoryProvider):
    async def get(self, user_id: str, key: str) -> Any:
        ...

    async def set(self, user_id: str, key: str, value: Any) -> None:
        ...

    async def search(self, user_id: str, query: str) -> list[dict]:
        ...