PydanticAI + Signbee: Type-Safe Python Agents for E-Signatures (2026)

import asyncio
import os
from dataclasses import dataclass
from decimal import Decimal
from typing import Literal

import httpx
from pydantic import BaseModel, EmailStr, Field, model_validator
from pydantic_ai import Agent, ModelRetry, RunContext


# 1. Dependency Injection Container (Never exposed to LLM prompt)
@dataclass
class SignbeeDeps:
    http_client: httpx.AsyncClient
    signbee_api_key: str
    sender_name: str
    sender_email: str
    webhook_url: str


# 2. Strict Pydantic v2 Domain Models for Contract Clauses & Fees
class MilestoneItem(BaseModel):
    phase_name: str = Field(..., min_length=3, description="Name of project phase")
    deliverable: str = Field(..., min_length=10, description="Concrete deliverable specification")
    due_date: str = Field(..., pattern=r"^\d{4}-\d{2}-\d{2}$", description="ISO YYYY-MM-DD date")
    fee_usd: Decimal = Field(..., gt=Decimal("0"), le=Decimal("250000"))


class ContractSubmission(BaseModel):
    title: str = Field(..., min_length=5, max_length=140)
    counterparty_company: str = Field(..., min_length=2)
    recipient_name: str = Field(..., min_length=2)
    recipient_email: EmailStr
    governing_law: Literal[
        "England and Wales",
        "State of Delaware, USA",
        "State of New York, USA",
    ]
    payment_terms_days: Literal[14, 30, 45] = 14
    milestones: list[MilestoneItem] = Field(..., min_length=1, max_length=10)
    total_fee_usd: Decimal = Field(..., gt=Decimal("0"))
    confidentiality_years: int = Field(default=3, ge=1, le=10)

    @model_validator(mode="after")
    def verify_milestone_sum(self) -> "ContractSubmission":
        calculated = sum((m.fee_usd for m in self.milestones), Decimal("0"))
        if calculated != self.total_fee_usd:
            raise ValueError(
                f"Milestone sum ({calculated}) does not match total_fee_usd ({self.total_fee_usd})."
            )
        return self

    def to_commonmark(self, sender_name: str) -> str:
        rows = "\n".join(
            f"| {m.phase_name} | {m.deliverable} | {m.due_date} | ${m.fee_usd:,.2f} |"
            for m in self.milestones
        )
        return (
            f"# {self.title}\n\n"
            f"This Statement of Work is entered into between **{sender_name}** (\"Provider\") "
            f"and **{self.counterparty_company}** (\"Client\").\n\n"
            f"## 1. Milestone & Fee Schedule\n\n"
            f"| Phase | Deliverable | Target Date | Fee (USD) |\n"
            f"|---|---|---|---|\n"
            f"{rows}\n\n"
            f"**Total Contract Value:** ${self.total_fee_usd:,.2f} USD (Net {self.payment_terms_days} days).\n\n"
            f"## 2. Confidentiality & IP Assignment\n"
            f"Both parties agree to maintain strict confidentiality for {self.confidentiality_years} years. "
            f"Upon full payment of ${self.total_fee_usd:,.2f}, Provider assigns all deliverable IP to Client.\n\n"
            f"## 3. Governing Law & Cryptographic Execution\n"
            f"This Agreement is governed by the laws of **{self.governing_law}** and executed via "
            f"Signbee with an immutable SHA-256 audit certificate."
        )


# 3. Structured Result Returned to Application Layer
class DispatchedEnvelopeResult(BaseModel):
    document_id: str
    status: str
    recipient_email: EmailStr
    total_fee_usd: Decimal
    verification_url: str


# 4. Initialize Type-Safe PydanticAI Agent
contract_agent = Agent(
    "openai:gpt-4o",
    deps_type=SignbeeDeps,
    output_type=DispatchedEnvelopeResult,
    system_prompt=(
        "You are an autonomous legal engineering agent. Extract commercial terms accurately, "
        "verify that milestone fees sum exactly to total_fee_usd, and dispatch the contract "
        "for electronic signature using the send_contract_via_signbee tool."
    ),
)


@contract_agent.tool
async def send_contract_via_signbee(
    ctx: RunContext[SignbeeDeps],
    contract: ContractSubmission,
) -> dict:
    """Compiles a validated ContractSubmission into CommonMark and dispatches via Signbee API v1."""
    markdown_payload = contract.to_commonmark(sender_name=ctx.deps.sender_name)

    response = await ctx.deps.http_client.post(
        "https://signb.ee/api/v1/send",
        headers={
            "Authorization": f"Bearer {ctx.deps.signbee_api_key}",
            "Content-Type": "application/json",
        },
        json={
            "title": contract.title,
            "markdown": markdown_payload,
            "sender_name": ctx.deps.sender_name,
            "sender_email": ctx.deps.sender_email,
            "recipient_name": contract.recipient_name,
            "recipient_email": str(contract.recipient_email),
            "expires_in_days": 14,
            "webhook_url": ctx.deps.webhook_url,
        },
    )

    if response.status_code == 422 or response.status_code == 400:
        raise ModelRetry(f"Signbee API rejected payload: {response.text}")

    response.raise_for_status()
    data = response.json()
    doc_id = data["id"]
    return {
        "document_id": doc_id,
        "status": data.get("status", "PENDING"),
        "recipient_email": str(contract.recipient_email),
        "total_fee_usd": str(contract.total_fee_usd),
        "verification_url": f"https://signb.ee/verify/{doc_id}",
    }

import hashlib
import hmac
import os
from datetime import datetime
from typing import Literal

from fastapi import FastAPI, Header, HTTPException, Request
from pydantic import BaseModel, EmailStr

app = FastAPI(title="Signbee Webhook Receiver")
WEBHOOK_SECRET = os.environ["SIGNBEE_WEBHOOK_SECRET"].encode("utf-8")


class SignbeeWebhookPayload(BaseModel):
    event: Literal["document.signed", "document.viewed", "document.expired", "document.declined"]
    document_id: str
    title: str
    status: str
    recipient_email: EmailStr
    sha256_hash: str | None = None
    signed_pdf_url: str | None = None
    completed_at: datetime | None = None


@app.post("/api/webhooks/signbee")
async def handle_signbee_webhook(
    request: Request,
    x_signbee_signature: str = Header(..., alias="X-Signbee-Signature"),
):
    raw_body = await request.body()

    # Compute expected HMAC-SHA256 digest over exact raw request bytes
    expected_mac = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    provided_mac = x_signbee_signature.removeprefix("sha256=")

    if not hmac.compare_digest(expected_mac, provided_mac):
        raise HTTPException(status_code=401, detail="Invalid X-Signbee-Signature HMAC digest")

    # Parse into strict Pydantic v2 model after cryptographic verification
    event_data = SignbeeWebhookPayload.model_validate_json(raw_body)

    if event_data.event == "document.signed":
        # Trigger downstream CRM stage transition, Stripe invoice, or Slack alert
        print(
            f"[VERIFIED] Contract {event_data.document_id} signed by {event_data.recipient_email} | "
            f"SHA-256: {event_data.sha256_hash}"
        )

    return {"received": True, "document_id": event_data.document_id}

import asyncio
import os
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStdio

signbee_mcp_server = MCPServerStdio(
    command="npx",
    args=["-y", "signbee-mcp"],
    env={"SIGNBEE_API_KEY": os.environ["SIGNBEE_API_KEY"]},
)

agent = Agent(
    "anthropic:claude-sonnet-4-5",
    mcp_servers=[signbee_mcp_server],
    system_prompt="Draft clean CommonMark agreements and send them via Signbee MCP tools.",
)

async def main():
    async with agent.run_mcp_servers():
        result = await agent.run(
            "Send a mutual NDA between B2bee Ltd (Michael Beckett, michael@signb.ee) "
            "and Northstar AI (Liam Chen, liam@northstar.ai) governed by England and Wales law."
        )
        print(result.output)

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