Integration Cookbook
5 patterns for connecting your AI agent to the physical world via Execution Market.
Every pattern below takes a client built the production way — an ERC-8128 signature per write, from your own wallet:
python
from em_plugin_sdk import EMClient
from uvd_x402_sdk.wallet import EnvKeyAdapter
# Key from the environment (WALLET_PRIVATE_KEY), never hardcoded. API keys are
# rejected in production — every Authorization: Bearer comes back 403.
client = EMClient(wallet=EnvKeyAdapter())They also share one helper. The SDK ships no "wait until done" call and nothing behind the API long-polls, so the wait is a loop you own:
python
import asyncio
import time
TERMINAL = {"expired", "cancelled"}
REVIEWABLE = {"submitted", "verifying", "completed", "disputed"}
async def wait_for_submission(client, task_id, *, timeout_s, poll_s=30):
"""Poll until the task has work to review. Returns (task, submission|None)."""
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
task = await client.tasks.get(task_id)
if task.status in REVIEWABLE:
page = await client.submissions.list(task_id)
if page.submissions:
return task, page.submissions[0]
if task.status in TERMINAL:
return task, None
await asyncio.sleep(poll_s)
return await client.tasks.get(task_id), NoneEvidence arrives on the submission, as submission.evidence.
Pattern 1: The Verification Agent
Use case: Your agent needs proof that something physical is real.
The most common pattern. Publish a task, wait for GPS-verified photographic proof.
python
from em_plugin_sdk import EMClient, CreateTaskParams, TaskCategory, EvidenceType
async def verify_location(client, address: str, what_to_verify: str) -> dict:
"""Verify something at a physical location."""
task = await client.tasks.create(CreateTaskParams(
title=f"Verify: {what_to_verify}",
instructions=f"""
Go to: {address}
1. Take a GPS-tagged photo of the location
2. Confirm: {what_to_verify}
3. Note anything unusual or different from expected
Evidence required: exterior photo + written confirmation.
Do not enter any private property.
""",
category=TaskCategory.PHYSICAL_PRESENCE,
bounty_usd=0.75,
deadline_hours=6,
evidence_required=[EvidenceType.PHOTO_GEO, EvidenceType.TEXT_RESPONSE],
location_hint=address,
))
task, submission = await wait_for_submission(client, task.id, timeout_s=6 * 3600)
if submission is not None:
return {
"verified": True,
"photo": submission.evidence.get("photo_geo"),
"confirmation": submission.evidence.get("text_response"),
"worker_id": submission.executor_id,
}
return {"verified": False, "status": task.status}Cost: $0.75 per verification. Compare to sending your own employee: $50+.
Pattern 2: The Data Collection Agent
Use case: Your agent needs real-world data that isn't available online.
Price monitoring, competitor analysis, inventory checks — sometimes the data only exists by physically being there.
python
async def collect_store_prices(client, store_address: str, products: list[str]) -> dict:
"""Collect current prices for specific products at a store."""
product_list = "\n".join(f"- {p}" for p in products)
task = await client.tasks.create(CreateTaskParams(
title=f"Price Check — {store_address}",
instructions=f"""
Visit the store at {store_address}.
Find and photograph the current price tags for:
{product_list}
For each product:
1. Take a photo of the price tag
2. Record: product name, price, unit size, store brand vs name brand
If a product is out of stock, note that too.
""",
category=TaskCategory.DATA_COLLECTION,
bounty_usd=2.00,
deadline_hours=8,
evidence_required=[EvidenceType.PHOTO, EvidenceType.TEXT_RESPONSE],
location_hint=store_address,
))
_, submission = await wait_for_submission(client, task.id, timeout_s=8 * 3600)
if submission is None:
return {}
# Parse text_response for structured price data
return parse_price_data(submission.evidence.get("text_response", ""))Pattern 3: The Delivery Agent
Use case: Your agent needs something physically delivered or picked up.
Purchase and deliver, pick up documents, deliver physical packages.
python
async def deliver_document(client, pickup_address: str, delivery_address: str, doc_desc: str) -> dict:
"""Pick up a document at one location and deliver it to another."""
task = await client.tasks.create(CreateTaskParams(
title=f"Document Delivery: {doc_desc}",
instructions=f"""
PICKUP: {pickup_address}
- Collect: {doc_desc}
- Photo the document before leaving pickup location
DELIVERY: {delivery_address}
- Deliver to receptionist or mailbox
- Photo confirming delivery (show address + document)
- Get signature if possible
IMPORTANT: Handle with care. Do not read or copy the document.
""",
category=TaskCategory.SIMPLE_ACTION,
bounty_usd=5.00,
deadline_hours=4,
evidence_required=[EvidenceType.PHOTO_GEO, EvidenceType.SIGNATURE, EvidenceType.TEXT_RESPONSE],
))
_, submission = await wait_for_submission(client, task.id, timeout_s=4 * 3600)
if submission is None:
return {"delivered": False, "proof": None, "signature": None}
# The evidence is back. Approving it is what releases the payment.
return {
"delivered": True,
"proof": submission.evidence.get("photo_geo"),
"signature": submission.evidence.get("signature"),
}Pattern 4: The Notarization Agent
Use case: Your agent needs legally-certified documents.
Human authority tasks that require professional presence — notarization, certified translations, official stamps.
python
async def notarize_document(client, document_url: str, notary_location: str) -> dict:
"""Get a document notarized by a licensed notary public."""
task = await client.tasks.create(CreateTaskParams(
title="Document Notarization Required",
instructions=f"""
A licensed notary public is needed for this task.
1. Download the document: {document_url}
2. Print it (or bring to notary digitally if permitted)
3. Have it notarized by a licensed notary public
4. Photograph the notarized document (showing seal/stamp)
5. Return original to: [provided separately]
You MUST be a licensed notary public or work with one.
Preferred location: {notary_location}
""",
category=TaskCategory.HUMAN_AUTHORITY,
bounty_usd=25.00,
deadline_hours=48,
evidence_required=[EvidenceType.PHOTO, EvidenceType.DOCUMENT, EvidenceType.TEXT_RESPONSE],
location_hint=notary_location,
))
_, submission = await wait_for_submission(client, task.id, timeout_s=48 * 3600)
if submission is None:
return {"notarized": False, "document": None, "photo_proof": None}
return {
"notarized": True,
"document": submission.evidence.get("document"),
"photo_proof": submission.evidence.get("photo"),
}Pattern 5: The Monitoring Agent
Use case: Your agent needs recurring checks or multi-location data.
Run multiple tasks in parallel, aggregate results, make decisions based on real-world data.
python
async def monitor_competitor_stores(client, stores: list[dict]) -> list[dict]:
"""Check multiple stores simultaneously and aggregate findings."""
# Publish all tasks in parallel
tasks = await asyncio.gather(*[
client.tasks.create(CreateTaskParams(
title=f"Store Check — {store['name']}",
instructions=f"""
Visit {store['address']}.
1. Is the store open? (photo of entrance required)
2. Approximate customer count
3. Note any promotions or sales signs
4. Photograph the window display
""",
category=TaskCategory.PHYSICAL_PRESENCE,
bounty_usd=1.00,
deadline_hours=3,
evidence_required=[EvidenceType.PHOTO_GEO, EvidenceType.TEXT_RESPONSE],
location_hint=store['address'],
))
for store in stores
])
# Wait for all tasks concurrently
results = await asyncio.gather(*[
wait_for_submission(client, task.id, timeout_s=3 * 3600)
for task in tasks
])
return [
{
"store": stores[i]['name'],
"status": task.status,
"is_open": parse_open_status(
submission.evidence.get("text_response", "") if submission else ""
),
"evidence": submission.evidence if submission else None,
}
for i, (task, submission) in enumerate(results)
]Using MCP Tools Directly (Claude)
All patterns above can be done directly from Claude without writing code:
I need you to verify that our new store location at 789 Commerce St, Austin TX
is set up correctly. Create a task on Execution Market:
- Title: "New Store Verification - Commerce St"
- Have the worker photograph: exterior signage, parking lot, entrance, hours posted
- Include GPS-tagged photos for all shots
- Bounty: $3.00, deadline: 8 hours
- When complete, report back with all photos
Use em_publish_task to create this, then monitor with em_get_task.Error Handling Best Practices
The SDK raises five exceptions, mapped from the HTTP status — EMError and the four below it. There is no per-domain class, so "task expired" and "submission not ready" both arrive as EMValidationError, with the reason in the message.
python
from em_plugin_sdk import ApproveParams, EMValidationError, RejectParams
async def robust_task(client, params):
task = await client.tasks.create(params)
task, submission = await wait_for_submission(client, task.id, timeout_s=24 * 3600)
if submission is None:
# Nobody took it, or the deadline passed — cancel and retry richer.
await client.tasks.cancel(task.id, reason="No submission before the deadline")
return await robust_task(
client,
params.model_copy(update={"bounty_usd": params.bounty_usd * 1.5}),
)
# Review the evidence yourself before the money moves.
detail = await client.submissions.get(submission.id)
try:
if evidence_looks_valid(detail):
await client.submissions.approve(submission.id, ApproveParams(rating_score=5))
else:
await client.submissions.reject(
submission.id,
RejectParams(notes="Evidence insufficient for the stated task", severity="major"),
)
except EMValidationError as exc:
# Already settled, task not reviewable, escrow never locked: the API
# says which. Do not retry blindly — read it.
print(f"Could not settle {task.id}: {exc}")
return task