OpenAI Agents SDK + Signbee: Handoffs, Guardrails & Certified E-Signatures (2026)

import os
import httpx
from pydantic import BaseModel, Field
from agents import (
    Agent,
    Runner,
    function_tool,
    handoff,
    input_guardrail,
    output_guardrail,
    GuardrailFunctionOutput,
    RunContextWrapper,
    TResponseInputItem,
)

# 1. Define Structured Schema for Guardrail Evaluation & Handoff Payload
class DealComplianceAudit(BaseModel):
    is_compliant: bool = Field(description="True if discount <= 20% and corporate email is valid")
    discount_percentage: float = Field(description="Detected discount percentage (0 to 100)")
    violation_reason: str | None = Field(default=None, description="Explanation if non-compliant")

class ContractDispatchReceipt(BaseModel):
    document_id: str
    signing_url: str
    recipient_email: str
    governing_law_included: bool

# 2. Define the Specialized Guardrail Auditor Agent
compliance_auditor_agent = Agent(
    name="Commercial Policy Auditor",
    instructions=(
        "Inspect the user request for contract generation. "
        "Flag is_compliant=False if the user requests a discount greater than 20%, "
        "asks to waive limitation of liability, or provides a non-corporate recipient email."
    ),
    output_type=DealComplianceAudit,
)

@input_guardrail
async def commercial_policy_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    input_data: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    result = await Runner.run(compliance_auditor_agent, input_data, context=ctx.context)
    audit: DealComplianceAudit = result.final_output
    return GuardrailFunctionOutput(
        output_info=audit,
        tripwire_triggered=not audit.is_compliant,
    )

# 3. Define the Signbee E-Signature Function Tool (Zero-Coordinate JSON Dispatch)
@function_tool
async def send_contract_via_signbee(
    title: str,
    markdown_body: str,
    recipient_name: str,
    recipient_email: str,
    expires_in_days: int = 14,
) -> dict:
    """
    Compiles CommonMark Markdown into a paginated A4 PDF and dispatches a legally binding
    two-party e-signature ceremony via Signbee with a SHA-256 cryptographic certificate.
    """
    api_key = os.environ["SIGNBEE_API_KEY"]
    async with httpx.AsyncClient(timeout=15.0) as client:
        response = await client.post(
            "https://signb.ee/api/v1/send",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            json={
                "title": title,
                "markdown": markdown_body,
                "sender_name": "B2Bee Enterprise Legal Ops",
                "sender_email": "legal@b2bee.ai",
                "recipient_name": recipient_name,
                "recipient_email": recipient_email,
                "expires_in_days": expires_in_days,
            },
        )
        response.raise_for_status()
        return response.json()

# 4. Output Guardrail: Verify Governing Law & Valid Signbee Receipt
@output_guardrail
async def legal_receipt_output_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    output: ContractDispatchReceipt,
) -> GuardrailFunctionOutput:
    invalid_dispatch = (
        not output.document_id.startswith(("doc_", "cm"))
        or not output.governing_law_included
    )
    return GuardrailFunctionOutput(
        output_info=output,
        tripwire_triggered=invalid_dispatch,
    )

# 5. Define the Privileged Legal Closing Agent & Front-Line Triage Agent
legal_closing_agent = Agent(
    name="Legal Closing Agent",
    instructions=(
        "You are an autonomous Legal Closing Agent. Draft a complete Master Services Agreement "
        "in CommonMark Markdown containing a Markdown pricing table, mutual confidentiality, "
        "a 12-month fee liability cap, and England & Wales governing law. "
        "Then call send_contract_via_signbee to dispatch the agreement for e-signature."
    ),
    tools=[send_contract_via_signbee],
    output_type=ContractDispatchReceipt,
    output_guardrails=[legal_receipt_output_guardrail],
)

commercial_triage_agent = Agent(
    name="Commercial Triage Agent",
    instructions=(
        "You assist account executives with structuring B2B SaaS deals. "
        "Once the customer name, corporate email, seat count, and pricing are clear, "
        "hand off immediately to the Legal Closing Agent to draft and dispatch the contract."
    ),
    handoffs=[handoff(legal_closing_agent)],
    input_guardrails=[commercial_policy_guardrail],
)

import { Agent, run, tool } from "@openai/agents";
import { z } from "zod";

// 1. Define strongly-typed Signbee E-Signature Tool using Zod
export const signbeeSendContractTool = tool({
  name: "signbee_send_contract",
  description:
    "Dispatches a legally binding contract for two-party electronic signature via Signbee. " +
    "Provide either CommonMark markdown (auto-compiled to A4 PDF) or an existing pdf_url.",
  parameters: z.object({
    title: z.string().min(5).describe("Official contract title shown on the Signing Certificate"),
    markdown: z
      .string()
      .optional()
      .describe("Full CommonMark Markdown contract text including GFM pricing tables"),
    pdf_url: z
      .string()
      .url()
      .optional()
      .describe("Public HTTPS URL to a pre-rendered PDF agreement if not using Markdown"),
    sender_name: z.string().describe("Full legal name of Party A (Sender)"),
    sender_email: z.string().email().describe("Verified corporate email of Party A"),
    recipient_name: z.string().describe("Full legal name of Party B (Counterparty)"),
    recipient_email: z.string().email().describe("Corporate email address of Party B"),
    expires_in_days: z.number().int().min(1).max(90).default(14),
  }),
  execute: async (input) => {
    // Deterministic pre-dispatch assertion inside the tool boundary
    if (!input.markdown && !input.pdf_url) {
      throw new Error("Either 'markdown' or 'pdf_url' must be provided to dispatch a contract.");
    }

    const response = await fetch("https://signb.ee/api/v1/send", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${process.env.SIGNBEE_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify(input),
    });

    if (!response.ok) {
      const errorPayload = await response.text();
      throw new Error(`Signbee API dispatch failed (${response.status}): ${errorPayload}`);
    }

    const data = await response.json();
    return {
      status: "DISPATCHED",
      document_id: data.document_id,
      signing_url: data.signing_url,
      verification_ledger: `https://signb.ee/verify/${data.document_id}`,
    };
  },
});

// 2. Specialized Legal Closing Agent
export const legalClosingAgent = new Agent({
  name: "Legal Closing Specialist",
  instructions: `You are the Legal Closing Specialist. When handed a finalized deal:
1. Verify that both sender_email and recipient_email are valid corporate domains.
2. Draft a clean, comprehensive CommonMark agreement with a Markdown deliverable table,
   intellectual property assignment, mutual NDA, and governing law clause.
3. Call signbee_send_contract and return the document_id and verification_ledger URL.`,
  tools: [signbeeSendContractTool],
});

// 3. Front-Line Revenue Operations Triage Agent with Handoff
export const revOpsTriageAgent = new Agent({
  name: "RevOps Deal Desk Agent",
  instructions: `You are the RevOps Deal Desk Agent. Help sales reps structure SOWs and MSAs.
Never attempt to send contracts yourself. Once commercial terms (client name, recipient email,
deliverables, and total contract value) are confirmed, hand off to the Legal Closing Specialist.`,
  handoffs: [legalClosingAgent],
});

// 4. Execute the Multi-Agent Pipeline
export async function executeAutonomousDealDesk(prompt: string) {
  const result = await run(revOpsTriageAgent, prompt);
  return result.finalOutput;
}

import os
import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

async def run_mcp_legal_agent():
    # Spawn Signbee's official MCP server over stdio with your API key
    async with MCPServerStdio(
        name="Signbee E-Signature MCP Server",
        params={
            "command": "npx",
            "args": ["-y", "signbee-mcp"],
            "env": {
                **os.environ,
                "SIGNBEE_API_KEY": os.environ["SIGNBEE_API_KEY"],
            },
        },
    ) as signbee_mcp_server:
        # The agent automatically inherits send_document, get_document_status, and verify tools
        closing_agent = Agent(
            name="Autonomous MCP Contract Closer",
            instructions=(
                "Use the Signbee MCP server tools to draft and dispatch legally binding "
                "agreements in CommonMark Markdown and check document execution status."
            ),
            mcp_servers=[signbee_mcp_server],
        )

        result = await Runner.run(
            closing_agent,
            "Draft and send a Mutual Non-Disclosure Agreement between B2bee Ltd (Michael Beckett, "
            "michael@b2bee.ai) and Northstar Robotics Ltd (Liam O'Connor, liam@northstar-robotics.io)."
        )
        print(result.final_output)

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