Example: programmatic integration

Scenario

You are building your own agent rather than using a desktop client. You want Titan’s tools available to your model, with production-grade handling for polling, retries, and errors.

Install an SDK

npm install @modelcontextprotocol/sdk

pip install mcp

Connect

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const transport = new StreamableHTTPClientTransport(
  new URL("https://mcp.webscraping.titannet.io/mcp"),
  {
    requestInit: {
      headers: { Authorization: `Bearer ${process.env.TITAN_API_KEY}` },
    },
  },
);

const client = new Client({ name: "my-agent", version: "1.0.0" });
await client.connect(transport);

const { tools } = await client.listTools();
console.log(tools.map((t) => t.name));

import os
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "https://mcp.webscraping.titannet.io/mcp"
HEADERS = {"Authorization": f"Bearer {os.environ['TITAN_API_KEY']}"}

async with streamablehttp_client(URL, headers=HEADERS) as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await session.list_tools()
        print([t.name for t in tools.tools])

A resilient call wrapper

Wrap tool calls once so every call gets the same error handling, retries, and polling.

const TERMINAL = new Set(["completed", "partial", "failed", "cancelled"]);

async function callTool(name: string, args: Record<string, unknown>) {
  const result = await client.callTool({ name, arguments: args });
  const body = JSON.parse(result.content[0].text);

  if (result.isError) {
    const err = new Error(`${body.code}: ${body.message}`);
    (err as any).retryable = body.retryable;
    (err as any).code = body.code;
    throw err;
  }

  return body;
}

async function callAndWait(
  name: string,
  args: Record<string, unknown>,
  { pollMs = 3000, maxWaitMs = 300_000 } = {},
) {
  let body = await callTool(name, args);
  const deadline = Date.now() + maxWaitMs;

  while (body.run_id && !TERMINAL.has(body.status)) {
    if (Date.now() > deadline) {
      throw new Error(`Run ${body.run_id} did not finish within ${maxWaitMs}ms`);
    }
    await new Promise((r) => setTimeout(r, pollMs));
    body = await callTool("titan_get_run", { run_id: body.run_id });
  }

  return body;
}

async function withRetry<T>(fn: () => Promise<T>, maxAttempts = 3): Promise<T> {
  for (let attempt = 1; ; attempt++) {
    try {
      return await fn();
    } catch (err: any) {
      if (!err.retryable || attempt >= maxAttempts) throw err;
      await new Promise((r) => setTimeout(r, 1000 * 2 ** (attempt - 1)));
    }
  }
}

import asyncio
import json
import time

TERMINAL = {"completed", "partial", "failed", "cancelled"}


class TitanToolError(Exception):
    def __init__(self, code, message, retryable):
        super().__init__(f"{code}: {message}")
        self.code = code
        self.retryable = retryable


async def call_tool(session, name, args):
    result = await session.call_tool(name, args)
    body = json.loads(result.content[0].text)

    if result.isError:
        raise TitanToolError(body["code"], body["message"], body.get("retryable", False))

    return body


async def call_and_wait(session, name, args, poll_s=3, max_wait_s=300):
    body = await call_tool(session, name, args)
    deadline = time.monotonic() + max_wait_s

    while body.get("run_id") and body.get("status") not in TERMINAL:
        if time.monotonic() > deadline:
            raise TimeoutError(f"Run {body['run_id']} did not finish in {max_wait_s}s")
        await asyncio.sleep(poll_s)
        body = await call_tool(session, "titan_get_run", {"run_id": body["run_id"]})

    return body


async def with_retry(coro_factory, max_attempts=3):
    for attempt in range(1, max_attempts + 1):
        try:
            return await coro_factory()
        except TitanToolError as err:
            if not err.retryable or attempt == max_attempts:
                raise
            await asyncio.sleep(2 ** (attempt - 1))

callAndWait collapses the synchronous and asynchronous cases into one path, so calling code does not branch on whether a run finished inside the wait window.

Expose the tools to a model

MCP tool definitions map onto the Claude API tool format directly:

import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();
const { tools } = await client.listTools();

const claudeTools = tools.map((t) => ({
  name: t.name,
  description: t.description,
  input_schema: t.inputSchema,
}));

async function runAgent(question: string) {
  const messages: Anthropic.MessageParam[] = [{ role: "user", content: question }];

  while (true) {
    const response = await anthropic.messages.create({
      model: "claude-sonnet-5",
      max_tokens: 4096,
      tools: claudeTools,
      messages,
    });

    messages.push({ role: "assistant", content: response.content });

    if (response.stop_reason !== "tool_use") {
      return response.content.find((b) => b.type === "text")?.text ?? "";
    }

    const toolResults = [];
    for (const block of response.content) {
      if (block.type !== "tool_use") continue;

      try {
        const output = await withRetry(() =>
          callAndWait(block.name, block.input as Record<string, unknown>),
        );
        toolResults.push({
          type: "tool_result" as const,
          tool_use_id: block.id,
          content: JSON.stringify(output),
        });
      } catch (err: any) {
        toolResults.push({
          type: "tool_result" as const,
          tool_use_id: block.id,
          content: err.message,
          is_error: true,
        });
      }
    }

    messages.push({ role: "user", content: toolResults });
  }
}

Returning tool errors to the model as is_error results lets it adapt—switching search providers after a rate limit, or narrowing a query that returned nothing.

Several agent frameworks connect to MCP servers natively, which removes the loop above entirely. Check whether yours supports remote MCP before writing your own.

Add idempotency

Derive a stable key from the request so retries never double-charge:

import { createHash } from "node:crypto";

function idempotencyKey(sessionId: string, name: string, args: unknown) {
  const digest = createHash("sha256")
    .update(JSON.stringify({ name, args }))
    .digest("hex")
    .slice(0, 16);
  return `${sessionId}-${digest}`;
}

const output = await callAndWait(name, {
  ...args,
  idempotency_key: idempotencyKey(sessionId, name, args),
});

See Idempotency and retries.

Reconcile spend

Every run carries a run_id that is also the Titan execution_id, so agent cost is auditable:

const spend = await fetch(
  `${process.env.TITAN_API_URL}/api/v1/billing/usage/executions/${runId}/spend-summary`,
  { headers: { Authorization: `Bearer ${process.env.TITAN_API_KEY}` } },
).then((r) => r.json());

You can also read usage.credits_consumed from each tool response and accumulate it per session, which avoids an extra API call when you only need a running total.

Production checklist

Concern	Approach
Keys	Load from a secret manager; one key per agent
Scopes	Grant only the tools the agent uses
Polling	Back off; never poll in a tight loop
Retries	Branch on `retryable`; cap attempts; use idempotency keys
Timeouts	Bound total wait per run and surface a clear failure
Cost	Track `usage.credits_consumed` per session and cap it
Observability	Log `request_id` and `run_id` on every call

Next steps

Research agent — a worked search-and-fetch pipeline
Errors and warnings — every code your handler will see
Connect your client — desktop and editor clients instead

Get started

Tool reference

How it works

Integration examples

Scenario

Install an SDK

Connect

A resilient call wrapper

Expose the tools to a model

Add idempotency

Reconcile spend

Production checklist

Next steps

​Scenario

​Install an SDK

​Connect

​A resilient call wrapper

​Expose the tools to a model

​Add idempotency

​Reconcile spend

​Production checklist

​Next steps

Scenario

Install an SDK

Connect

A resilient call wrapper

Expose the tools to a model

Add idempotency

Reconcile spend

Production checklist

Next steps