How to let your AI agent hire a human in 5 minutes (MCP + Python)

{
  "mcpServers": {
    "invoke-human-tasks": {
      "type": "streamable-http",
      "url": "https://invoke.nanocorp.app/mcp"
    }
  }
}

python -m venv .venv
source .venv/bin/activate
pip install "mcp>=1.13.0" "langchain-core>=0.3.0"

import asyncio
import json
from typing import Any

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

INVOKE_MCP_URL = "https://invoke.nanocorp.app/mcp"


def extract_structured_result(result: Any) -> dict[str, Any]:
    if getattr(result, "is_error", False):
        message = "Invoke MCP post_task failed"
        if getattr(result, "content", None):
            message = getattr(result.content[0], "text", message)
        raise RuntimeError(message)

    structured = getattr(result, "structured_content", None)
    if isinstance(structured, dict):
        return structured

    if getattr(result, "content", None):
        text = getattr(result.content[0], "text", "")
        if text:
            return json.loads(text)

    raise RuntimeError(f"Unexpected Invoke MCP result: {result!r}")


async def post_task(payload: dict[str, Any]) -> dict[str, Any]:
    async with streamable_http_client(INVOKE_MCP_URL) as streams:
        read_stream, write_stream, *_ = streams
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            result = await session.call_tool("post_task", arguments=payload)
            return extract_structured_result(result)


async def main() -> None:
    response = await post_task({
        "title": "Call two local bakeries about gluten-free cupcakes",
        "description": (
            "Call two bakeries in downtown Austin. Ask whether they can provide "
            "12 gluten-free cupcakes tomorrow. Return price, pickup window, "
            "phone number called, and employee name if available."
        ),
        "reward_cents": 1500,
        "contact_email": "you@example.com",
        "category": "phone_research",
    })

    print(json.dumps(response, indent=2))


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

import asyncio
from typing import Optional

from langchain_core.tools import StructuredTool

from invoke_post_task import post_task


def hire_human(
    title: str,
    description: str,
    reward_cents: int,
    contact_email: str,
    category: Optional[str] = None,
) -> dict:
    """Hire a human when the agent hits work it cannot complete itself."""
    payload = {
        "title": title,
        "description": description,
        "reward_cents": reward_cents,
        "contact_email": contact_email,
    }
    if category:
        payload["category"] = category

    return asyncio.run(post_task(payload))


hire_human_tool = StructuredTool.from_function(
    func=hire_human,
    name="hire_human",
    description=(
        "Post a task to Invoke for real-world work an AI agent cannot do: "
        "phone calls, local checks, verification, judgment, or hands-on tasks. "
        "The first post for a poster email is published free."
    ),
)

SYSTEM:
You are an AI agent with a hire_human tool.

Use hire_human instead of guessing when the next step requires:
- calling a business or person
- visiting or checking a physical place
- verifying an account, code, listing, form, or website by hand
- human judgment, taste, or policy review
- evidence such as photos, screenshots, call notes, timestamps, or source links

When calling hire_human, write a bounded worker brief with:
1. exact outcome needed
2. context and links the worker needs
3. reward_cents between 1000 and 50000
4. acceptance criteria
5. proof required from the worker