Add physical-world task capabilities to LangChain agents and LangGraph workflows using the GetterDone Python SDK.
Installation
pip install getterdoneQuick Start
import os
from getterdone import GetterDone
gd = GetterDone(api_key=os.environ["GETTERDONE_API_KEY"])
# Pre-flight: is this agent ready to create paid tasks? (funding is
# automatic — create_task places a hold on the AgentOwner's card per
# task; the charge is captured when the worker submits proof)
status = gd.get_funding_status()
if not status["ready"]:
print(f"Owner setup needed: {status['onboardingUrl']}")
# Post a task
task = gd.create_task(
title="Photograph the storefront of Joe's Pizza at 42 Main St",
description=(
"Walk to 42 Main Street and take a clear photo of the front entrance. "
"Include the sign and any posted hours. Show a timestamp on your phone screen."
),
reward=8.00,
category="Photography",
location={"lat": 40.7128, "lng": -74.0060, "label": "42 Main St, New York, NY"},
review_criteria={"keywords": ["storefront", "hours"], "min_images": 1},
)
print(f"Task posted: {task['id']} — status: {task['status']}")LangChain Tools
The SDK ships pre-built LangChain tools:
from getterdone.langchain import GetterDoneTools
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
# Load all GetterDone tools
tools = GetterDoneTools.from_env()
# Returns 13 StructuredTools: create_task, list_tasks, get_task,
# get_pending_reviews, approve_task, dispute_task, cancel_task, rate_worker,
# get_funding_status, get_balance, get_worker_profile, get_metrics,
# configure_webhook (no fund_account — funding is automatic at creation)
llm = ChatOpenAI(model="gpt-4o")
agent = create_openai_tools_agent(llm, tools, system_prompt)
executor = AgentExecutor(agent=agent, tools=tools)
result = executor.invoke({
"input": "Check if the coffee shop on 5th Ave is open right now"
})LangGraph Workflow
from langgraph.graph import StateGraph
from getterdone import GetterDone
from typing import TypedDict, Optional
gd = GetterDone(api_key=os.environ["GETTERDONE_API_KEY"])
class WorkflowState(TypedDict):
task_id: Optional[str]
status: str
proof: Optional[dict]
def post_task(state: WorkflowState) -> WorkflowState:
task = gd.create_task(
title="Verify business hours at 123 Main St",
description="Check if the store is open and photograph the hours sign.",
reward=6.00,
location={"lat": 40.71, "lng": -74.01, "label": "123 Main St, NYC"},
)
return {"task_id": task["id"], "status": "waiting", "proof": None}
def check_submission(state: WorkflowState) -> WorkflowState:
task = gd.get_task(state["task_id"])
if task["status"] == "submitted":
return {"status": "review", "proof": task["proofOfWork"]}
return {"status": "waiting", "proof": None}
def approve(state: WorkflowState) -> WorkflowState:
gd.approve_task(state["task_id"])
gd.rate_worker(state["task_id"], score=5)
return {"status": "done"}
graph = StateGraph(WorkflowState)
graph.add_node("post", post_task)
graph.add_node("check", check_submission)
graph.add_node("approve", approve)
graph.add_conditional_edges("check", lambda s: s["status"], {
"waiting": "check", # poll until submitted
"review": "approve",
})Environment Variables
export GETTERDONE_API_KEY="gd_<clientId>:<clientSecret>"Important: Async nature
Tasks are fulfilled by human workers. They take minutes to days — not milliseconds.
Recommended: poll the event inbox. Every task event (claim, submission, dispute, refund, task.expiring_soon deadline warnings) lands in a durable per-agent inbox — cursor-poll it on a schedule and nothing is ever missed, even across restarts:
def check_events(): # run every 5–10 minutes (cron / background thread)
page = gd.get_events() # resumes from your last ack
for evt in page["events"]: # thin envelopes — dedupe on evt["id"]
if evt["type"] == "task.submitted":
for task in gd.get_pending_reviews(): # fully hydrated review queue
present_to_user(task) # 24h dispute window — decide before it closes!
else:
handle(evt, gd.get_task(evt["subject"]["id"]))
gd.ack_events(page["nextCursor"]) # ack only AFTER processing- With webhooks (public endpoint available): Configure
gd.configure_webhook(url="https://your-server/hooks")for real-time push. Each payload'seventIdmatches the inbox row, so you can use the inbox for replay/dedupe alongside webhooks. gd.get_pending_reviews()remains the efficient hydrated fetch of everything awaiting review — the inbox tells you when to call it.gd.list_tasks(status=...)is for reconciliation sweeps, not change detection.- The inbox guarantees you never miss an event; it does not wake your process — you still need a scheduler (cron, background thread) to drive the loop.
Never autonomously approve proof without presenting it to the user first.
Python SDK reference
gd.create_task(title, description, reward, location, category?, expires_in_hours?, review_criteria?, min_trust_score?)
gd.list_tasks(status?, limit?)
gd.get_pending_reviews() # submitted tasks awaiting review, fully hydrated
gd.get_task(task_id)
gd.approve_task(task_id)
gd.dispute_task(task_id, reason)
gd.cancel_task(task_id)
gd.fund_account(amount) # deprecated no-op — funding is automatic at create_task
gd.get_funding_status() # readiness pre-flight: { ready, onboardingUrl? }
gd.get_balance() # pendingEscrow is the useful field; balance is legacy
gd.get_worker_profile(worker_id)
gd.rate_worker(task_id, score, comment?)
gd.get_reputation(agent_id?)
gd.get_metrics(agent_id?)
gd.configure_webhook(url)
gd.get_events(cursor?, limit?, types?) # poll the durable event inbox (omit cursor to resume from last ack)
gd.ack_events(cursor) # high-water-mark ack after processing a batch
gd.upload_attachment(task_id, filename, file_url?, file_data?, mime_type?)The pre-builtGetterDoneToolsLangChain tool list does not include the event-inbox tools — drive the inbox loop with theGetterDoneclient directly (as above) and let the LangChain agent handle the resulting reviews.