> ## Documentation Index
> Fetch the complete documentation index at: https://webscraping.titannet.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Example: programmatic integration

> Connect the Titan MCP server from your own agent runtime in TypeScript or Python, wire the tools into an LLM loop, and handle runs, retries, and errors.

## 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

<CodeGroup>
  ```bash TypeScript theme={null}
  npm install @modelcontextprotocol/sdk
  ```

  ```bash Python theme={null}
  pip install mcp
  ```
</CodeGroup>

## Connect

<CodeGroup>
  ```typescript TypeScript theme={null}
  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));
  ```

  ```python Python theme={null}
  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])
  ```
</CodeGroup>

## A resilient call wrapper

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

<CodeGroup>
  ```typescript TypeScript theme={null}
  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)));
      }
    }
  }
  ```

  ```python Python theme={null}
  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))
  ```
</CodeGroup>

`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:

```typescript theme={null}
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.

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

## Add idempotency

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

```typescript theme={null}
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](/docs/mcp/idempotency-and-retries).

## Reconcile spend

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

```typescript theme={null}
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](/docs/mcp/examples/research-agent) — a worked search-and-fetch pipeline
* [Errors and warnings](/docs/mcp/errors-and-warnings) — every code your handler will see
* [Connect your client](/docs/mcp/connect-your-client) — desktop and editor clients instead