Integrations

npx @impri/mcp
# env: IMPRI_API_KEY=im_...  IMPRI_BASE_URL=http://localhost:8484

{
  "mcpServers": {
    "impri": {
      "command": "npx",
      "args": ["@impri/mcp"],
      "env": {
        "IMPRI_API_KEY": "im_<your-key>",
        "IMPRI_BASE_URL": "http://localhost:8484"
      }
    }
  }
}

"IMPRI_BASE_URL": "https://api.impri.dev"

from langchain.tools import BaseTool
from impri import ImpriClient, ImpriRejected

client = ImpriClient()  # reads IMPRI_API_KEY, IMPRI_BASE_URL from env

class ApprovedEmailTool(BaseTool):
    name = "send_email"
    description = "Send an email — requires human approval before sending."

    async def _arun(self, to: str, body: str) -> str:
        async with client.approval_gate(
            kind="email.send",
            title=f"Send email to {to}",
            preview={"format": "plain", "body": body},
            editable=["preview.body"],
            timeout_s=300,
        ) as approved:
            # Use the human-approved (possibly edited) body
            await mailer.send(to=to, body=approved.final_preview["body"])
        return "Sent."

async def approval_node(state: GraphState) -> GraphState:
    action = await client.create_action(
        kind=state["planned_action"],
        title=state["proposed_title"],
        preview={"format": "markdown", "body": state["proposed_body"]},
        editable=["preview.body"],
    )
    # store for the next node; the graph can be interrupted here
    return {**state, "pending_action_id": action.id, "inbox_url": action.inbox_url}

async def execute_node(state: GraphState) -> GraphState:
    action = await client.get_action(state["pending_action_id"])
    if action.status == "approved":
        body = action.decision.final_preview["body"]
        await perform(body)
        await client.report_result(action.id, "executed")
    return state

from agents import FunctionTool
from impri import ImpriClient, ImpriRejected

client = ImpriClient()

@client.requires_approval(
    kind="code.exec",
    title=lambda code, **_: f"Execute: {code[:60]}",
    preview=lambda code, **_: {"format": "plain", "body": code},
    editable=["preview.body"],
)
async def run_python(code: str) -> str:
    return subprocess.check_output(["python", "-c", code], text=True)

tools = [FunctionTool(run_python)]

pip install 'impri-openai[openai-agents]'

from agents import Agent, Runner
from impri_openai import ImpriClient
from impri_openai.guardrail import make_guardrail

client = ImpriClient()                    # reads IMPRI_API_KEY from env

approval = make_guardrail(
    client,
    kind="agent.run",
    title="Approve this agent task",
    preview_from_input=True,              # show the user's input in the inbox card
    timeout_s=300,
)

agent = Agent(
    name="my-agent",
    instructions="You are a helpful assistant.",
    input_guardrails=[approval],
)
result = await Runner.run(agent, "Summarise my emails")

pip install impri-crewai
# or with crewai bundled as an extra:
pip install "impri-crewai[crewai]"

import os
from crewai import Agent, Crew, Task
from impri_crewai import ImpriClient, ImpriApprovalTool, ImpriRejected

client = ImpriClient(api_key=os.environ["IMPRI_API_KEY"])

approval_tool = ImpriApprovalTool(
    client=client,
    action_kind="email.send",   # dot-namespaced category for inbox filtering
    timeout_s=600,              # seconds to wait for human decision (default 300)
    editable=["preview.body"],  # fields the reviewer may edit before approving
)

agent = Agent(
    role="Marketing assistant",
    goal="Draft and send campaign emails with human sign-off.",
    backstory="You help marketing teams communicate with customers.",
    tools=[approval_tool],
)

task = Task(
    description=(
        "Draft a follow-up email to newsletter subscribers and send it "
        "after getting human approval."
    ),
    expected_output="Confirmation that the email was approved and sent.",
    agent=agent,
)

crew = Crew(agents=[agent], tasks=[task])
try:
    result = crew.kickoff()
except ImpriRejected as exc:
    print(f"Human rejected the action: {exc}")

import os
from crewai import Agent, Crew, Task
from impri_crewai import ImpriClient, ImpriApprovalCallback, ImpriRejected

client = ImpriClient(api_key=os.environ["IMPRI_API_KEY"])

gate = ImpriApprovalCallback(
    client,
    action_kind="agent.output",         # kind string for inbox categorisation
    timeout_s=300,                      # seconds to wait (default 300)
    title_prefix="Review agent draft",  # prepended to the auto-generated title
)

agent = Agent(
    role="Content writer",
    goal="Write blog posts.",
    backstory="...",
)

task = Task(
    description="Write a 500-word blog post about Impri.",
    expected_output="A complete blog post.",
    agent=agent,
)

crew = Crew(
    agents=[agent],
    tasks=[task],
    step_callback=gate,   # gate every intermediate step output
    # task_callback=gate  # alternative: gate only the final task output
)

try:
    result = crew.kickoff()
except ImpriRejected as exc:
    print(f"Step rejected by human reviewer: {exc}")

import { ImpriClient } from '@impri/sdk'
import { withImpriApproval } from '@impri/claude-agent-sdk'

const impri = new ImpriClient({ apiKey: process.env.IMPRI_API_KEY! })

const sendEmailGated = withImpriApproval({
  toolDef: {
    name: 'send_email',
    description: 'Send an email to a recipient.',
    input_schema: {
      type: 'object',
      properties: {
        to:   { type: 'string' },
        body: { type: 'string' },
      },
      required: ['to', 'body'],
    },
  },
  execute: async ({ to, body }) => {
    await emailService.send({ to: String(to), body: String(body) })
    return `Email sent to ${to}.`
  },
  impriClient: impri,
  kind: 'email.send',
  title: ({ to }) => `Send email to ${to}`,
  preview: ({ body }) => ({ format: 'plain', body: String(body) }),
  editable: ['preview.body'],
  onRejected: (err) => `Email rejected (action ${err.actionId}).`,
})

// In your agent loop:
const response = await anthropic.messages.create({
  model: 'claude-opus-4-5',
  tools: [sendEmailGated.toolDef],  // unchanged — Claude sees the normal tool
  messages,
})

for (const block of response.content) {
  if (block.type === 'tool_use' && block.name === 'send_email') {
    const result = await sendEmailGated.handle(block)
    // feed result back to Claude as tool_result
  }
}

{
  "kind": "{{1.kind}}",
  "title": "{{1.title}}",
  "preview": { "format": "plain", "body": "{{1.body}}" },
  "callback_url": "https://hook.make.com/<your-scenario-webhook>",
  "expires_in": 86400
}

// Express / Fastify / Hono (TypeScript)
import { verifyWebhook, ImpriWebhookSignatureError } from '@impri/sdk'

app.post('/impri/webhook', express.raw({ type: 'application/json' }), async (req, res) => {
  try {
    verifyWebhook(
      req.body,
      process.env.IMPRI_WEBHOOK_SECRET!,
      req.headers['x-impri-timestamp'] as string,
      req.headers['x-impri-nonce'] as string,
      req.headers['x-impri-signature'] as string,
    )
  } catch (e) {
    if (e instanceof ImpriWebhookSignatureError) return res.status(400).end()
    throw e
  }

  const event = JSON.parse(req.body.toString())
  if (event.status === 'approved') {
    await enqueue({ actionId: event.action_id, body: event.final_preview?.body })
  }
  res.status(200).end()
})

# FastAPI / Flask / Django (Python)
import impri
from impri import ImpriWebhookSignatureError

@app.post("/impri/webhook")
async def webhook(request: Request):
    raw = await request.body()
    try:
        impri.verify_webhook(
            raw_body=raw,
            secret=os.environ["IMPRI_WEBHOOK_SECRET"],
            timestamp=request.headers["X-Impri-Timestamp"],
            nonce=request.headers["X-Impri-Nonce"],
            signature=request.headers["X-Impri-Signature"],
        )
    except ImpriWebhookSignatureError:
        raise HTTPException(status_code=400)

    event = await request.json()
    if event["status"] == "approved":
        await enqueue(action_id=event["action_id"], body=event["final_preview"]["body"])
    return Response(status_code=200)