Embedding Duvo as a Callable Tool in Your AI System

Register Duvo Assignments as callable tools your own AI agent or copilot invokes through the Duvo MCP server — your agent orchestrates, Duvo handles the operational task.

If you’re building your own AI agent, copilot, or workflow runner, you can register individual Duvo Assignments as callable tools that your agent invokes when its work enters Duvo’s domain. Your agent stays in control of the overall workflow — Duvo handles the specific operational task and hands the result back.

When to embed Duvo vs. build the capability yourself

Embed Duvo when…	Build natively when…
The task already exists as a configured Duvo Assignment	The task is simple and doesn’t need connections or HITL
The task involves approvals, sensitive systems, or audit needs	Latency is critical and a Job round-trip isn’t acceptable
Non-developers need to maintain the task’s SOP	You want full control over the execution environment
The task spans multiple Connections (Gmail + Sheets + Slack…)	The task has no human-in-the-loop requirement

Common patterns:

• An editor-agent reviews a document and delegates any compliance action to a Duvo Assignment.
• A customer copilot triages a request and hands a regulated branch (refund, escalation) to Duvo, where a human can approve it.
• An internal AI workbench catalogs Duvo Assignments as named skills and routes work to them by category.

How it works

Your agent connects to the Duvo MCP server (https://api.duvo.ai/v2/mcp). Every Duvo Public API endpoint is auto-exposed as an MCP tool. Your agent calls those tools to start Jobs, poll for completion, respond to human-in-the-loop requests, and collect results — all through a standard MCP interface.

Your agent
  └─► Duvo MCP server (https://api.duvo.ai/v2/mcp)
        └─► Duvo Assignment (Job runs, Connections, HITL)
              └─► Result returned to your agent

For a guide on connecting your MCP host to the Duvo MCP server and authenticating, see The Duvo MCP server.

Scoping which Assignments your agent can call

The Duvo MCP server exposes tools for the full Public API. Your agent can call listAgents to discover all Assignments visible to its API key, then filter by name or ID to invoke specific ones.

Narrowing scope with API keys:

Each API key is scoped to a team and inherits the permissions of the user who created it. To limit a parent agent to a subset of Assignments, create a dedicated Duvo user with access only to the relevant Assignments and generate an API key for that user. This prevents the parent agent from accidentally discovering or starting Assignments it shouldn’t touch.

Discovering Assignments at startup:

# MCP tool call — exact syntax depends on your MCP client library
tools_result = mcp_client.call_tool("listAgents", {"limit": 50})
assignments = tools_result["agents"]

How an Assignment appears as a tool

When your agent calls listAgents, each Assignment is returned with metadata your agent can use to decide which one to invoke. Here is a trimmed example of the response (additional fields omitted for brevity):

{
  "agents": [
    {
      "id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
      "name": "Process Refund Request",
      "team_id": "string",
      "created_at": "string",
      "updated_at": "string",
      "last_run_at": "string",
      "created_by": { "id": "string", "name": "string", "email": "string" },
      "latest_build": {
        "id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
        "agent_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
        "name": "string",
        "revision_name": "string",
        "revision_description": "string",
        "revision_number": 0,
        "status": "string"
      },
      "integration_configs": [
        {
          "integration_id": "string",
          "integration_type": "string",
          "integration_name": "string",
          "is_connected": true
        }
      ]
    }
  ],
  "total": 0,
  "limit": 50,
  "offset": 0
}

The key fields your agent needs are id, name, and latest_build.revision_description. The name and description come from what was entered in the Duvo dashboard — edit them there to make the Assignment self-describing for your agent.

Tips for writing agent-friendly Assignment metadata:

• Name: use a verb phrase — “Process Refund Request”, “Triage Support Ticket”, “Update Inventory Record”.
• Revision description (latest_build.revision_description): explain the input the Assignment expects and the output it produces — “Takes a customer complaint email body. Returns a triage decision (escalate / auto-resolve / needs-more-info) with a one-sentence rationale.”

The Assignment’s SOP governs what the agent actually does. The name and description only affect how clearly your parent agent can decide when to call it.

Invocation patterns

All Job endpoints are on the base URL https://api.duvo.ai/v2. Starting a Job is team-scoped (POST /teams/{team_id}/runs); reading and responding to a Job is addressed by run_id.

Synchronous (poll until done)

Start a Job, then poll getRun until the status is completed, failed, or stopped. Suitable for short-running Assignments (under a few minutes).

import time
import httpx

BASE_URL = "https://api.duvo.ai/v2"
API_KEY  = "dv_your_key_here"
TEAM_ID  = "your-team-id"

headers = {"Authorization": f"Bearer {API_KEY}"}

# 1. Start the Job
run_resp = httpx.post(
    f"{BASE_URL}/teams/{TEAM_ID}/runs",
    headers=headers,
    json={
        "agent_id": "your-assignment-id",
        "message":  "Process refund request for order #99182. Customer says item never arrived.",
    },
)
run_id = run_resp.json()["run"]["id"]

# 2. Poll for completion
while True:
    status_resp = httpx.get(f"{BASE_URL}/runs/{run_id}", headers=headers)
    status = status_resp.json()["run"]["status"]

    if status in ("completed", "failed", "stopped"):
        break

    time.sleep(5)

# 3. Retrieve messages
messages_resp = httpx.get(f"{BASE_URL}/runs/{run_id}/messages", headers=headers)
result = messages_resp.json()["messages"]

Asynchronous with a webhook

Provide a webhook_url when starting the Job. Duvo POSTs to that URL on state changes (run_completed, run_failed, run_interrupted) and when the Assignment needs input (human_request), so your agent doesn’t have to poll. Filter on the payload’s event field for the case you care about.

run_resp = httpx.post(
    f"{BASE_URL}/teams/{TEAM_ID}/runs",
    headers=headers,
    json={
        "agent_id":    "your-assignment-id",
        "message":     "...",
        "webhook_url": "https://your-agent.example.com/duvo/events",
    },
)

A human-input event looks like this:

{
  "event": "human_request_created",
  "run_id": "550e8400-...",
  "request_id": "req_789xyz",
  "title": "Confirm data deletion",
  "description": "About to delete 150 records. Confirm?"
}

Respond when ready:

httpx.post(
    f"{BASE_URL}/runs/{run_id}/human-requests/{request_id}/respond",
    headers=headers,
    json={"approved": True},
)

HITL handoff

When Duvo reaches a step that requires human approval, the Job’s status changes to waiting. If your parent agent is polling, it detects this status and can either:

• Forward the request to a human — surface the title and description from getRun’s pending_human_request field in your agent’s own UI or chat thread, then relay the human’s answer back via respondToHumanRequest.
• Respond programmatically — if your agent has enough context to make the decision itself, it can call respondToHumanRequest directly without surfacing it to a human.

# Detect waiting status during poll
if status == "waiting":
    # The pending HITL request is embedded in the getRun response
    run = httpx.get(f"{BASE_URL}/runs/{run_id}", headers=headers).json()["run"]
    req = run.get("pending_human_request")
    if req:
        # Surface to your user / decide programmatically
        approved = ask_operator(req["title"], req["description"])

        httpx.post(
            f"{BASE_URL}/runs/{run_id}/human-requests/{req['id']}/respond",
            headers=headers,
            json={"approved": approved},
        )

Trust and guardrails

What the parent agent can’t do:

• The parent agent operates under the permissions of its API key. It cannot bypass Connection-level authorization — if the Assignment uses a Gmail Connection that the API key’s user can’t access, the Job will fail with a permissions error.
• The parent agent cannot modify the Assignment’s SOP or Connections through an MCP tool call without the corresponding write endpoints, which should be locked down for service accounts.

Audit trail:

Every Job started via the API is recorded with the API key’s user identity in the Duvo audit log. If multiple parent agents share the same key, their Jobs are indistinguishable. Create separate API keys per parent agent to maintain a clean audit trail. See Audit Log and Activity Tracking for how to export and query the log.

High-risk actions:

Assignments that take irreversible actions (send emails, delete records, submit transactions) should have Human-in-the-Loop gates in their SOP. A parent agent that responds to HITL requests programmatically bypasses those gates — only do this if your agent has verified the action is safe. See Guardrails for High-Risk Automations for the full risk framework.

Rate limits:

All Public API rate limits apply to MCP tool calls: 5,000 requests per minute per API key. Long-polling loops should still sleep between polls (5–30 seconds) to avoid burning through quota during busy periods.

End-to-end example: email triage with a human approval gate

A parent agent monitors an inbound email queue. When it receives a complaint that touches a financial policy, it delegates to a Duvo Assignment, waits for a human to approve the proposed response, then sends the final email.

import time
import httpx

BASE_URL = "https://api.duvo.ai/v2"
API_KEY  = "dv_your_key_here"
TEAM_ID  = "your-team-id"
TRIAGE_ASSIGNMENT_ID = "assign_triage_abc123"

headers = {"Authorization": f"Bearer {API_KEY}"}

def handle_complaint(email_body: str) -> str:
    """
    Parent agent detects a financial complaint, delegates to Duvo,
    surfaces the HITL approval, and returns the final outcome.
    """

    # 1. Start the Job — Duvo will draft a response and request approval
    run_resp = httpx.post(
        f"{BASE_URL}/teams/{TEAM_ID}/runs",
        headers=headers,
        json={
            "agent_id": TRIAGE_ASSIGNMENT_ID,
            "message":  f"Inbound complaint: {email_body}",
        },
    )
    run_id = run_resp.json()["run"]["id"]
    print(f"Job started: {run_id}")

    # 2. Poll for completion or HITL pause
    while True:
        run = httpx.get(f"{BASE_URL}/runs/{run_id}", headers=headers).json()["run"]
        status = run["status"]

        if status == "completed":
            # Retrieve final messages from Duvo
            msgs = httpx.get(f"{BASE_URL}/runs/{run_id}/messages", headers=headers)
            return msgs.json()["messages"][-1]["text_content"]

        if status in ("failed", "stopped"):
            raise RuntimeError(f"Job ended with status: {status}")

        if status == "waiting":
            # 3. Surface the HITL request to an operator
            req = run.get("pending_human_request")
            if req:
                print("\n--- Approval needed ---")
                print(f"Title:   {req['title']}")
                print(f"Details: {req['description']}")
                decision = input("Approve? (y/n): ")

                httpx.post(
                    f"{BASE_URL}/runs/{run_id}/human-requests/{req['id']}/respond",
                    headers=headers,
                    json={"approved": decision.lower() == "y"},
                )

        time.sleep(5)


result = handle_complaint(
    "I was charged twice for my order #99182. I need a refund immediately."
)
print(f"\nOutcome: {result}")

What happens step by step:

1. The parent agent starts a Job on the “Complaint Triage” Assignment with the raw email body.
2. Duvo’s Assignment reads the complaint, connects to the order system via its configured Connection, drafts a refund proposal, and pauses — asking an operator to approve before sending.
3. The parent agent detects waiting, fetches the HITL request, and surfaces the draft to an on-call operator (via a chat message, Slack notification, or UI).
4. The operator approves or rejects. The parent agent relays that decision back to Duvo.
5. Duvo sends the approved response and marks the Job completed.
6. The parent agent reads the final result and continues its own workflow.

Related

• The Duvo MCP server — how to configure authentication and connect any MCP-compatible host
• Connect a custom MCP to Duvo — the inverse direction: bring your own tools into Duvo Assignments
• Running Assignments via API — full API reference for Jobs, messages, and HITL responses
• Guardrails for High-Risk Automations — risk framework for delegating irreversible actions
• Audit Log and Activity Tracking — how to trace which agent started which Job