Give a Claude Agent SDK Agent a Send-Text Tool

A custom tool is four parts passed to the tool() helper in TypeScript or the @tool decorator in Python: a name Claude uses to call it, a description Claude reads to decide when to call it, an input schema for its arguments, and an async handler that runs when Claude calls it (Agent SDK custom tools). The description earns its keep here: "sends a text message" tells Claude nothing about when a second call is wrong, while "call this once per intended message, and only reuse idempotency_key when retrying the exact same send" gives the model the rule it needs.

Here's a Python version that keeps the dedup check inside the handler, ahead of the Beam call:

import os
from typing import Any
import httpx
from claude_agent_sdk import tool, create_sdk_mcp_server

SENT_KEYS: dict[str, dict[str, Any]] = {}  # swap for a real datastore before this touches production

@tool(
    "send_text",
    "Send one real text message to one phone number through Beam. Call this once per "
    "intended message; reuse the same idempotency_key only to retry that exact send.",
    {
        "type": "object",
        "properties": {
            "to": {"type": "string", "description": "Recipient phone number, E.164 or 10-digit US"},
            "message": {"type": "string", "description": "The exact text to send"},
            "idempotency_key": {
                "type": "string",
                "description": "A stable id for this specific send, such as 'reminder-482-day1'.",
            },
        },
        "required": ["to", "message", "idempotency_key"],
    },
)
async def send_text(args: dict[str, Any]) -> dict[str, Any]:
    key = args["idempotency_key"]
    if key in SENT_KEYS:
        cached = SENT_KEYS[key]
        return {"content": [{"type": "text", "text": f"Already sent, Beam id {cached['id']}. Not sending again."}]}

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "https://beam.aisync.link/v1/messages",
            headers={"x-api-key": os.environ["BEAM_API_KEY"]},
            json={"to": args["to"], "message": args["message"]},
        )
    if resp.status_code != 200:
        return {"content": [{"type": "text", "text": f"Beam rejected the send: {resp.text}"}], "is_error": True}

    data = resp.json()
    SENT_KEYS[key] = data  # record before returning so a crash right here still blocks the next duplicate attempt
    return {"content": [{"type": "text", "text": f"Sent. Beam id {data['id']}, status {data['status']}."}]}

beam_server = create_sdk_mcp_server(name="beam", version="1.0.0", tools=[send_text])

Register the server and the tool's fully qualified name, mcp__{server_name}__{tool_name}, with ClaudeAgentOptions before calling query():

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        mcp_servers={"beam": beam_server},
        allowed_tools=["mcp__beam__send_text"],
    )
    async for message in query(
        prompt="Text the saved test contact to confirm tomorrow's 2pm visit, idempotency key confirm-482-day1.",
        options=options,
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

asyncio.run(main())

Diagram: a box labeled "your process" containing a "send_text handler" box connected to a small "idempotency store" box, with one arrow out to "Beam POST /v1/messages" and that arrow continuing to a phone icon, plus a dashed arrow showing a repeat call being caught by the store before it ever reaches Beam.

The handler's error path matters as much as its happy path. An uncaught exception becomes an SDK error result carrying the raw message, and Claude can see that and try again; a handler that catches the failure and returns is_error: True with a composed message gives Claude a clearer reason, here "already sent" instead of a stack trace (Agent SDK custom tools). Since the agent loop can retry on an error result, that composed message does real work: the difference between Claude retrying blindly and Claude knowing the first attempt already succeeded.

Because Beam's plain text-send endpoint doesn't give you one. POST /v1/messages accepts to, message, and a few optional fields for attachments and a sender name; nothing in that body tells Beam "this is the same send as before." Beam does support that pattern elsewhere: a request_key field its MCP tool and email send endpoint both require, where reusing the key returns the original result instead of repeating it. The REST endpoint a self-hosted send_text tool calls isn't one of those surfaces, so calling it twice with the same number and text gets two separate texts, each with its own message id, each going out.

That gap has to be filled in your own handler, and the schema above is where it starts. Requiring idempotency_key forces every call, whether from the model, a retry, or a test script, to carry a stable label checked against its own store before it reaches Beam. Where that key comes from matters too: a timestamp generated fresh on every attempt defeats the point, since a genuine retry would then look like a new send. A key tied to the actual business event, a specific appointment or follow-up step, is what makes a repeat recognizable as a repeat.

It's tempting to assume the SDK's idempotentHint annotation already covers this. It doesn't: the docs describe annotations as metadata, not enforcement, and idempotentHint specifically as informational, something tooling can read, not something that blocks a second call from running your handler's code (Agent SDK custom tools). Set it if accurate, since a correctly built send_text handler is genuinely safe to call twice with the same arguments, but keep it alongside the real dedup check, not instead of it.

Run this against a number you own before send_text ever sees a real contact list.

1. Set BEAM_API_KEY to a key from a workspace with no production contacts, and set to on every test call to your own number.
2. Call query() with a prompt that should trigger exactly one tool call, and confirm the agent's transcript shows one tool_use block for mcp__beam__send_text, not two.
3. Confirm the one text you expected arrives on your phone, and check Beam's own message record for that id rather than trusting the handler's return text alone.
4. Call the handler again with the identical idempotency_key from step two, outside of any agent run, and confirm it returns the cached result with no second text arriving.
5. Temporarily make the handler raise an exception right after the Beam call succeeds but before it returns, then let Claude retry with the same key, and confirm it returns the cached result instead of sending a second text.
6. Omit idempotency_key from a manual tool call and confirm the schema rejects it outright, rather than the handler silently treating a missing key as a new send.
7. If you've wired a permission_policy or a canUseTool gate, confirm a send attempt pauses for approval before it reaches Beam, not just before the model's final answer.
8. Grep whatever you log from this run for the literal value of BEAM_API_KEY and confirm it never appears: a key inside a schema field would show up here, a key read from the environment inside the handler won't.
9. Only after steps one through eight pass clean, point the same tool at a small slice of real contacts, not the full list.

That list covers the tool's own correctness. For the broader pre-launch pass, delivery confirmation beyond the queued response, rate limit behavior under a burst, and a dry run against STOP handling, our pre-launch testing guide walks through the fuller sequence before any Claude texting agent touches a live number.