Build an agent

sw run t-0e4243 --agent claude-code -- sh -c 'claude -p "$STATEWEAVE_PROMPT" \
  --mcp-config "$STATEWEAVE_MCP_CONFIG" --strict-mcp-config \
  --allowedTools "mcp__slack__*"'

pip install stateweave-sdk "anthropic[mcp]" zenpy
export ANTHROPIC_API_KEY=sk-ant-...      # from console.anthropic.com

"""A complete agent for StateWeave: Claude with the official MCP servers of the run's apps (Slack, Gmail, Salesforce,
Shopify), whose tools the vendors define, plus two tools of its own for Zendesk, which has no MCP server.

    pip install "anthropic[mcp]" zenpy
    export ANTHROPIC_API_KEY=sk-ant-...
    sw run t-753b29 --agent my-agent@0.1 -- python my_agent.py

`sw run` hands it the run (STATEWEAVE_RUN_FILE, STATEWEAVE_PROMPT, ...). For a question, the answer goes to
STATEWEAVE_ANSWER_FILE. An app with neither an MCP server nor tools here is out of the agent's reach."""
import asyncio
import contextlib
import itertools
import json
import os
import sys

import anthropic
import httpx2
from anthropic import beta_async_tool
from anthropic.lib.tools.mcp import async_mcp_tool
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from zenpy import Zenpy
from zenpy.lib.api_objects import Comment

run = json.load(open(os.environ["STATEWEAVE_RUN_FILE"]))
answer = {}  # set by submit_answer


# ------------------------------------------------------------------ Zendesk: no MCP server, so two tools of our own
def zendesk() -> Zenpy:
    app = run["apps"]["zendesk"]
    subdomain, _, domain = app["base_url"].removeprefix("https://").partition(".")
    token = next(v for label, v in app["credentials"].items() if label.startswith("user:"))
    return Zenpy(subdomain=subdomain, domain=domain, oauth_token=token)


@beta_async_tool
async def find_zendesk_tickets(query: str) -> str:
    """Find Zendesk tickets, each with its people and comments. Uses Zendesk's search syntax: words from the ticket,
    a ticket number, status:open, tags:vip, ...

    Args:
        query: The search query.
    """
    def find():
        z = zendesk()
        return json.dumps([{
            "id": t.id, "subject": t.subject, "status": t.status, "priority": t.priority, "tags": t.tags,
            "requester": t.requester and t.requester.email, "assignee": t.assignee and t.assignee.email,
            "comments": [{"author": c.author.name, "public": c.public, "body": c.body[:2000]}
                         for c in z.tickets.comments(ticket=t.id)],
        } for t in itertools.islice(z.search(query, type="ticket"), 10)])
    return await asyncio.to_thread(find)


@beta_async_tool
async def update_zendesk_ticket(ticket_id: int, status: str | None = None, assignee_email: str | None = None,
                                add_tags: list[str] | None = None, comment: str | None = None,
                                public: bool = False) -> str:
    """Change a Zendesk ticket. Only what you give changes.

    Args:
        ticket_id: The ticket's number.
        status: new, open, pending, hold, solved or closed.
        assignee_email: The email of the agent to assign it to.
        add_tags: Tags to add.
        comment: A comment to add.
        public: True for a reply the requester sees, False for an internal note (the default).
    """
    def update():
        z = zendesk()
        t = z.tickets(id=ticket_id)
        if status:
            t.status = status
        if assignee_email:
            t.assignee_id = next(u.id for u in z.search(assignee_email, type="user") if u.email == assignee_email)
        if add_tags:
            t.tags = sorted(set(t.tags) | set(add_tags))
        if comment:
            t.comment = Comment(body=comment, public=public)
        z.tickets.update(t)
        return f"Ticket {ticket_id} updated."
    return await asyncio.to_thread(update)


@beta_async_tool
async def submit_answer(answer_json: str) -> str:
    """Hand in the answer to a question, as JSON in the format the task asks for.

    Args:
        answer_json: The answer as JSON text.
    """
    answer["value"] = json.loads(answer_json)
    return "Answer recorded."


# ------------------------------------------------------------------ the agent
async def main() -> None:
    async with contextlib.AsyncExitStack() as stack:
        tools = [submit_answer]
        for name, app in run["apps"].items():
            if app.get("mcp"):  # the vendor's MCP server: its tools, as the vendor defines them
                http = await stack.enter_async_context(httpx2.AsyncClient(headers=app["mcp"]["headers"]))
                streams = await stack.enter_async_context(streamable_http_client(app["mcp"]["url"], http_client=http))
                session = await stack.enter_async_context(ClientSession(streams[0], streams[1]))
                await session.initialize()
                tools += [async_mcp_tool(tool, session) for tool in (await session.list_tools()).tools]
            elif name == "zendesk":
                tools += [find_zendesk_tickets, update_zendesk_ticket]

        persona = run["persona"]
        system = (f"You work for {persona['name']} ({persona['title']}) and act with their accounts in the company's "
                  f"apps, using your tools. The apps in this task: {', '.join(run['apps'])}. If the task needs an app "
                  "you have no tool for, do what you can and say what you could not do. Look things up before you "
                  "change anything, and change only what the task asks for. "
                  + (f"When you know the answer, call submit_answer with it in this format: {run['answer_format']}"
                     if run["kind"] == "answer" else "When the changes are made, say in a sentence what you did."))

        client = anthropic.AsyncAnthropic()
        runner = client.beta.messages.tool_runner(
            model="claude-opus-5-5",
            max_tokens=16000,
            output_config={"effort": "high"},
            cache_control={"type": "ephemeral"},  # each turn resends the conversation so far: cache it
            betas=["server-side-fallback-2026-07-01"],
            fallbacks="default",  # if Claude Opus 5.5 declines, another model carries on
            max_iterations=60,
            system=system,
            tools=tools,
            messages=[{"role": "user", "content": run["prompt"]}],
        )
        async for message in runner:
            for block in message.content:
                if block.type == "tool_use":
                    print(f"  {block.name} {json.dumps(block.input)[:120]}", file=sys.stderr)
        if message.stop_reason == "refusal":
            print("Claude declined the task", file=sys.stderr)

    if "value" in answer:
        with open(os.environ["STATEWEAVE_ANSWER_FILE"], "w") as f:
            json.dump(answer["value"], f)


asyncio.run(main())

sw run t-72abae --agent my-agent@0.1 -- python my_agent.py

sw: run r-ef8604ff of t-72abae started: https://acme.app.stateweave.net/#/runs/r-ef8604ff
  list-orders {"query": "Harbor", "first": 20}
  find {"q": "FIND {Harbor} IN ALL FIELDS RETURNING Account(Id, Name), Contact(Id, Name, Email, AccountId)"}
  list-customers {"query": "email:simone.hartley@harbor-co.example"}
  soqlQuery {"q": "SELECT Id, Name, Description, ShippingStreet, ShippingCity, ShippingState, ShippingPostalCode, …"}
  …
  graphql_mutation {"query": "mutation($input: OrderInput!) { orderUpdate(input: $input) { order { id note shipping…"}
  updateSobjectRecord {"sobject-name": "Note", "id": "002iu0000Buxt3sAQA", "body": {"Body": "Account ACC-0142. Order…"}}
passed  23 calls  https://acme.app.stateweave.net/#/runs/r-ef8604ff
  NG-4817 shipping address and note and the Harbor & Co. delivery note now carry Pier 9 / 410 Waterfront Ave by noon on 2026-11-06; nothing else changed