Name: Agentfield
Rating: 5 (1 reviews)

Suspend agent execution at a specific point and wait for a human decision before continuing. The control plane transitions the execution to a "waiting" state, and the agent resumes automatically when the external approval service posts a webhook callback.

Serving Required

app.pause() requires the agent to be serving (app.serve()) because the control plane needs a callback URL to notify the agent when the approval resolves. Calling pause() without serving raises RuntimeError.

Basic Example

from agentfield import Agent
from agentfield.client import AgentFieldClientError

app = Agent(node_id="loan-processor")

@app.reasoner()
async def process_loan_application(application_id: str, amount: float) -> dict:
    """Process a loan application with human underwriter approval."""

    # Run automated checks first
    credit_score = await run_credit_check(application_id)
    risk_score = await calculate_risk(application_id, amount)

    # Pause and wait for underwriter decision
    result = await app.pause(
        approval_request_id=f"loan-{application_id}",
        approval_request_url=f"https://approvals.example.com/loans/{application_id}",
        expires_in_hours=48,
    )

    if result.approved:
        return {
            "status": "approved",
            "application_id": application_id,
            "amount": amount,
            "reviewer_notes": result.feedback,
        }
    else:
        return {
            "status": "rejected",
            "application_id": application_id,
            "reason": result.feedback,
        }

`app.pause()`

Signature

async def pause(
    self,
    approval_request_id: str,
    approval_request_url: str = "",
    expires_in_hours: int = 72,
    timeout: Optional[float] = None,
    execution_id: Optional[str] = None,
) -> ApprovalResult

Parameters

Prop

Type

Returns: ApprovalResult - The human reviewer's decision, including feedback and the raw webhook payload.

`app.wait_for_resume()`

Signature

async def wait_for_resume(
    self,
    approval_request_id: str,
    execution_id: Optional[str] = None,
    timeout: Optional[float] = None,
) -> ApprovalResult

Use wait_for_resume() for crash recovery scenarios. Unlike app.pause(), it does not call the control plane to register a new approval request. It simply registers a local asyncio.Future keyed by approval_request_id and waits for the webhook callback to arrive. If timeout elapses before the callback arrives, it falls back to a single status poll against the control plane.

Parameters

Prop

Type

Returns: ApprovalResult - The human reviewer's decision retrieved either from the webhook callback or the status poll fallback.

`ApprovalResult`

The dataclass returned by both app.pause() and app.wait_for_resume().

from dataclasses import dataclass
from typing import Any, Dict, Optional

@dataclass
class ApprovalResult:
    decision: str
    feedback: str = ""
    execution_id: str = ""
    approval_request_id: str = ""
    raw_response: Optional[Dict[str, Any]] = None

    @property
    def approved(self) -> bool:
        return self.decision == "approved"

    @property
    def changes_requested(self) -> bool:
        return self.decision == "request_changes"

Fields

Prop

Type

Properties

Property	Type	Description
`approved`	`bool`	`True` when `decision == "approved"`
`changes_requested`	`bool`	`True` when `decision == "request_changes"`

How It Works Under the Hood

Common Patterns

Handling Rejection with Feedback

from agentfield import Agent
from agentfield.client import AgentFieldClientError

app = Agent(node_id="contract-reviewer")

@app.reasoner()
async def review_contract(contract_id: str, contract_text: str) -> dict:
    """Submit a contract for legal review and handle all decision outcomes."""

    app.note(
        f"## Contract Submitted for Review\n\n**Contract ID**: {contract_id}\n**Status**: Awaiting legal approval",
        tags=["approval", "waiting"],
    )

    result = await app.pause(
        approval_request_id=f"contract-{contract_id}",
        approval_request_url=f"https://legal.example.com/contracts/{contract_id}",
        expires_in_hours=24,
    )

    if result.approved:
        app.note(
            f"## Contract Approved\n\n**Reviewer Notes**: {result.feedback or 'None'}",
            tags=["approval", "approved"],
        )
        return {"status": "approved", "contract_id": contract_id}

    if result.changes_requested:
        app.note(
            f"## Changes Requested\n\n**Feedback**: {result.feedback}",
            tags=["approval", "changes-requested"],
        )
        return {
            "status": "changes_requested",
            "contract_id": contract_id,
            "required_changes": result.feedback,
        }

    if result.decision == "rejected":
        app.note(
            f"## Contract Rejected\n\n**Reason**: {result.feedback}",
            tags=["approval", "rejected"],
        )
        return {
            "status": "rejected",
            "contract_id": contract_id,
            "reason": result.feedback,
        }

    # Expired or error
    app.note(
        f"## Approval Request {result.decision.title()}\n\nNo decision was made within the deadline.",
        tags=["approval", result.decision],
    )
    return {"status": result.decision, "contract_id": contract_id}

Timeout Handling

import asyncio
from agentfield import Agent

app = Agent(node_id="expense-approver")

@app.reasoner()
async def approve_expense(expense_id: str, amount: float, employee_id: str) -> dict:
    """Approve an expense report with a local timeout fallback."""

    try:
        result = await app.pause(
            approval_request_id=f"expense-{expense_id}",
            approval_request_url=f"https://hr.example.com/expenses/{expense_id}",
            expires_in_hours=72,
            timeout=3600.0,  # Wait up to 1 hour for the webhook before polling
        )
    except asyncio.TimeoutError:
        # The local timeout elapsed and the status poll also failed
        return {
            "status": "timeout",
            "expense_id": expense_id,
            "message": "Approval request timed out. Please resubmit.",
        }

    return {
        "status": result.decision,
        "expense_id": expense_id,
        "amount": amount,
        "feedback": result.feedback,
    }

Crash Recovery with `wait_for_resume()`

If your agent process restarts while an approval is still pending, use wait_for_resume() to reconnect to the existing request without creating a duplicate.

from agentfield import Agent

app = Agent(node_id="trade-executor")

@app.reasoner()
async def execute_trade(trade_id: str, symbol: str, quantity: int) -> dict:
    """Execute a trade with compliance approval, resumable after crash."""

    approval_request_id = f"trade-{trade_id}"

    # Check if an approval is already in flight from a previous run
    existing = await check_pending_approval(trade_id)

    if existing:
        # Reconnect to the existing approval request instead of creating a new one
        app.note(
            f"## Resuming Pending Approval\n\n**Trade ID**: {trade_id}\n**Request ID**: {approval_request_id}",
            tags=["approval", "resuming"],
        )
        result = await app.wait_for_resume(
            approval_request_id=approval_request_id,
            timeout=300.0,
        )
    else:
        # First run - register a new approval request
        app.note(
            f"## Submitting Trade for Compliance Approval\n\n**Symbol**: {symbol}\n**Quantity**: {quantity}",
            tags=["approval", "waiting"],
        )
        result = await app.pause(
            approval_request_id=approval_request_id,
            approval_request_url=f"https://compliance.example.com/trades/{trade_id}",
            expires_in_hours=4,
        )

    if result.approved:
        return {"status": "approved", "trade_id": trade_id, "symbol": symbol}
    else:
        return {"status": result.decision, "trade_id": trade_id, "reason": result.feedback}

Error Handling with `AgentFieldClientError`

from agentfield import Agent
from agentfield.client import AgentFieldClientError

app = Agent(node_id="payment-authorizer")

@app.reasoner()
async def authorize_payment(payment_id: str, amount: float) -> dict:
    """Authorize a high-value payment with error handling."""

    try:
        result = await app.pause(
            approval_request_id=f"payment-{payment_id}",
            approval_request_url=f"https://payments.example.com/authorize/{payment_id}",
            expires_in_hours=2,
        )
    except AgentFieldClientError as e:
        # Control plane rejected the request (e.g., invalid execution state)
        return {
            "status": "error",
            "payment_id": payment_id,
            "error": f"Control plane error: {str(e)}",
        }
    except RuntimeError as e:
        # Agent is not serving - callback URL unavailable
        return {
            "status": "error",
            "payment_id": payment_id,
            "error": f"Agent not serving: {str(e)}",
        }

    return {
        "status": result.decision,
        "payment_id": payment_id,
        "amount": amount,
        "authorized": result.approved,
        "notes": result.feedback,
    }

Using `app.pause()` Inside a Complex Workflow

Combine app.pause() with app.call() and app.note() for multi-step workflows that require human checkpoints.

from agentfield import Agent

app = Agent(node_id="mortgage-orchestrator")

@app.reasoner()
async def process_mortgage_application(
    application_id: str,
    applicant_id: str,
    loan_amount: float,
    property_address: str,
) -> dict:
    """Full mortgage processing pipeline with human approval checkpoints."""

    app.note(
        f"## Mortgage Application Started\n\n**Application**: {application_id}\n**Amount**: ${loan_amount:,.2f}\n**Property**: {property_address}",
        tags=["mortgage", "start"],
    )

    # Step 1: Automated underwriting via specialized agent
    underwriting = await app.call(
        "underwriting-agent.run_automated_checks",
        applicant_id=applicant_id,
        loan_amount=loan_amount,
    )

    app.note(
        f"## Automated Underwriting Complete\n\n**Credit Score**: {underwriting.get('credit_score')}\n**DTI Ratio**: {underwriting.get('dti_ratio')}\n**Automated Decision**: {underwriting.get('recommendation')}",
        tags=["mortgage", "underwriting"],
    )

    # Step 2: Property appraisal via specialized agent
    appraisal = await app.call(
        "appraisal-agent.order_appraisal",
        property_address=property_address,
        loan_amount=loan_amount,
    )

    app.note(
        f"## Appraisal Received\n\n**Appraised Value**: ${appraisal.get('value'):,.2f}\n**LTV Ratio**: {appraisal.get('ltv_ratio')}",
        tags=["mortgage", "appraisal"],
    )

    # Step 3: Human underwriter approval for final decision
    app.note(
        "## Awaiting Human Underwriter Approval\n\nAll automated checks complete. Routing to senior underwriter.",
        tags=["mortgage", "approval", "waiting"],
    )

    approval = await app.pause(
        approval_request_id=f"mortgage-{application_id}",
        approval_request_url=f"https://underwriting.example.com/applications/{application_id}",
        expires_in_hours=48,
    )

    app.note(
        f"## Underwriter Decision Received\n\n**Decision**: {approval.decision.upper()}\n**Notes**: {approval.feedback or 'None'}",
        tags=["mortgage", "approval", "decision"],
    )

    if not approval.approved:
        return {
            "status": approval.decision,
            "application_id": application_id,
            "reason": approval.feedback,
        }

    # Step 4: Generate closing documents via document agent
    documents = await app.call(
        "document-agent.generate_closing_package",
        application_id=application_id,
        loan_amount=loan_amount,
        underwriting_data=underwriting,
        appraisal_data=appraisal,
    )

    app.note(
        f"## Closing Package Ready\n\n**Documents Generated**: {len(documents.get('files', []))}\n**Closing Date**: {documents.get('closing_date')}",
        tags=["mortgage", "closing", "complete"],
    )

    return {
        "status": "approved",
        "application_id": application_id,
        "loan_amount": loan_amount,
        "closing_date": documents.get("closing_date"),
        "document_package_id": documents.get("package_id"),
        "underwriter_notes": approval.feedback,
    }

Performance Considerations

Execution State:

While paused, the execution holds no CPU or memory resources on the agent process. The asyncio.Future is lightweight and the agent can handle other requests normally.

Webhook Latency:

Resume latency depends entirely on how quickly the external approval service posts the webhook. The SDK adds no meaningful overhead once the callback arrives.

Timeout vs. Expiry:

timeout is a local parameter controlling how long the SDK waits for the webhook before falling back to a status poll. expires_in_hours is stored on the control plane and determines when the approval request itself becomes invalid. Set timeout shorter than expires_in_hours to ensure the fallback poll can still retrieve a valid result.

Crash Recovery:

If the agent restarts while an approval is pending, use wait_for_resume() on startup to reconnect. The control plane retains the approval state across agent restarts.

Human-in-the-Loop - Conceptual overview of approval workflows in Agentfield
app.note() - Track approval state with real-time progress notes
Agent Execution REST API - REST endpoints for approval requests and webhook callbacks

app.pause()

app.pause()

Step-by-step callback flow