npm · mcp-ts-core0.9.7

Evidence

- Tool / resource / prompt **descriptions** templated from runtime data? Static strings are safer; templated descriptions enable "tool poisoning" (adversarial metadata steering the LLM toward a dangerous tool).

- Descriptions mutated mid-session? Rug-pull surface: client approved the v1 description, server now advertises v2 behavior.

**Smell:** `return { body: await fetch(url).then(r => r.text()) }` rendered directly in `format()`. Or: `description: \`Look up ${tenant.customLabel}\`` where `cus

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

async handler(input, ctx) {

  if (queue.full()) throw rateLimited('Queue at capacity');

  const items = await fetch(input.ids);

  if (items.length === 0) throw notFound(`No items match ${input.ids.length} IDs`);

  return { items };

}

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

// wire so format()-only clients see the hint mirrored into content[] text.

    if (queue.full()) throw ctx.fail('queue_full', undefined, { ...ctx.recoveryFor('queue_full') });

    const articles = await fetch(input.pmids);

    if (articles.length === 0) {

      // Dynamic recovery — interpolate runtime context, override the contract default.

      throw ctx.fail('no_pmid_match', `No data for ${input.pmids.length} PMIDs`, {

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

> **Declare contracts inline on each tool, even when similar across tools.** The contract is part of the tool's documented public surface — reading one tool definition file should give the full picture (input, output, errors, handler, format). Don't extract a shared `errors[]` constant or contract module to deduplicate near-identical entries; per-tool repetition is the intended cost of locality, and dynamic `recovery` hints often need tool-specific runtime context anyway. If a code-cleanup pass 

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

import { serviceUnavailable } from '@cyanheads/mcp-ts-core/errors';

export class NcbiService {

  async fetch(pmids: string[], ctx: Context) {

    const response = await fetchWithRetry(...);

    if (!response.ok) {

      throw serviceUnavailable(`NCBI returned HTTP ${response.status}`, {

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

output: z.object({ items: z.array(ItemSchema).describe('Resolved items') }),

  async handler(input, ctx) {

    if (queue.full()) throw ctx.fail('queue_full');

    const items = await fetch(input.ids);

    if (items.length === 0) throw ctx.fail('no_match', `No items match ${input.ids.length} IDs`, { ids: input.ids });

    // ctx.fail('typo')   ← TypeScript error: 'typo' isn't in the contract

    return { items };

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

#### Service-layer throws

API-wrapping tools usually delegate to a service: `await ncbi.fetch(input, ctx)`. The throw lives in the service, not the handler. Services accept `ctx` (the unified Context) so they can call `ctx.log`, `ctx.recoveryFor`, etc. The handler doesn't catch — it just bubbles, and the framework's auto-classifier preserves `data` on the wire.

The contract entry on the tool and the `data: { reason }` on the service throw need to use the **same reason string** so the two sides

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

],

  async handler(input, ctx) {

    const articles = await ncbi.fetch(input.pmids);

    if (articles.length === 0) {

      throw ctx.fail('no_match', `None of ${input.pmids.length} PMIDs returned data`);

    }

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

recovery: 'NCBI is degraded; retry in a few minutes.' },

  ],

  async handler(input, ctx) {

    return { articles: await ncbi.fetch(input.pmids, ctx) };  // throws bubble unchanged

  },

});

```

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

// Works with both async and sync functions

const result = await ErrorHandler.tryCatch(

  () => externalApi.fetch(url),

  {

    operation: 'ExternalApi.fetch',

    context: { url },

Remediation

Pass timeout= on every call: - HTTP: `requests.get(url, timeout=5)`, `httpx.get(url, timeout=5.0)` - Node fetch: `AbortSignal.timeout(5000)` - Subprocess: `subprocess.run(["cmd"], timeout=30, check=True)` Pick a value short enough to fail fast and retry.

Evidence

description: 'Search inventory items by query.',

  annotations: { readOnlyHint: true },

  input: z.object({

    query: z.string().describe('Search terms'),

    limit: z.number().default(10).describe('Max results'),

  }),

  output: z.object({

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

description: 'Search inventory items by query.',

  annotations: { readOnlyHint: true },

  input: z.object({

    query: z.string().describe('Search terms'),

    limit: z.number().default(10).describe('Max results'),

  }),

  output: z.object({

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

export const myTool = tool('my_tool', {

  description: 'Does something useful.',

  annotations: { readOnlyHint: true },

  input: z.object({ query: z.string().describe('Search query') }),

  output: z.object({

    items: z.array(z.object({

      id: z.string().describe('Item ID'),

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

export const reviewCode = prompt('review_code', {

  description: 'Review code for issues and best practices.',

  args: z.object({

    code: z.string().describe('Code to review'),

    language: z.string().optional().describe('Programming language'),

  }),

  generate: (args) => [

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

```typescript

output: z.object({

  path: z.string().describe('Resolved target path.'),

  created: z.boolean().describe('True when the operation created a new target.'),

  previousSizeInBytes: z.number().describe('Byte size before the mutation. Zero when created is true.'),

  currentSizeInBytes: z.number().describe('Byte size after the mutation. Equals previous when no-op.'),

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

```typescript

// Schema: keep permissive — accepts empty strings from form clients

input: z.object({

  query: z.string().describe('Search terms'),

  dateRange: z.object({

    minDate: z.string().describe('Start date (YYYY-MM-DD)'),

    maxDate: z.string().describe('End date (YYYY-MM-DD)'),

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

export const fetchAndStage = tool('fetch_and_stage_germplasm', {

  description: 'Fetch germplasm matching a query and stage it on a DataCanvas for follow-up SQL.',

  input: z.object({

    query: z.string().describe('Search query'),

    canvas_id: z

      .string()

      .optional()

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

export const reviewCode = prompt('review_code', {

  description: 'Review code for issues and best practices.',

  args: z.object({

    code: z.string().describe('Code to review'),

    language: z.string().optional().describe('Programming language'),

  }),

  generate: (args) => [

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

export const search = tool('search', {

  description: 'Search for items by query.',

  input: z.object({

    query: z.string().describe('Search query'),

    limit: z.number().default(10).describe('Max results'),

  }),

  output: z.object({

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

import { tool } from '@cyanheads/mcp-ts-core';

const myTool = tool('my_tool', {

  input: z.object({ query: z.string().describe('Search query') }),

  output: z.object({ result: z.string().describe('Search result') }),

  auth: ['tool:my_tool:read'],

  async handler(input, ctx) {

Remediation

Shape the schema to the tool's actual intent: - Zod: chain `.enum([...])`, `.regex(/.../)`, or `.max(n)`; prefer `z.enum([...])` or `z.literal(...)` when the value set is small. - Pydantic: use `Literal["a", "b"]` or `Field(max_length=..., pattern=r"...")`. - JSON Schema: add `"enum"`, `"pattern"`, or `"maxLength"` to the property. An overbroad schema is an "overpowered tool" — the model has nothing to prevent it from calling the tool with input far beyond what the tool's prose contract

Evidence

# README.md Conventions for MCP Servers

Structure and content guide for creating or updating a README for an MCP server built on `@cyanheads/mcp-ts-core`. If a README already exists, use this as a reference to audit and improve it — don't blindly rewrite sections that are already accurate.

## Structure

Use this section order. Omit sections that don't apply (e.g., skip Docker/Workers if the server doesn't deploy there).

```text

# {Server Name}                         ← centered HTML block (h1

Remediation

Declare a real authentication mechanism in the manifest, matching what the running server actually enforces: - `"auth": "bearer"` with a token scheme documented for callers - `"auth": "oauth"` / `"oauth2": { ... }` for delegated flows - `"apiKey": { "header": "X-API-Key", "prefix": "..." }` - `"mtls": true` when client certificates are required If the server is intentionally unauthenticated (stdio-only, local developer tool, trusted-host network), document the assumption in the manifest via a `"

Evidence

#!/usr/bin/env node

/**

 * @fileoverview Verifies that `skills/` (canonical) has been propagated to the

 * local mirrors `.agents/skills/` and `.claude/skills/`. The maintenance skill

 * updates `skills/` for downstream servers; the mirrors are what local agent

 * toolchains actually read, and silent drift means agents run on stale guidance.

 *

 * Propagation is one-way (`skills/` → mirrors), so only missing or

 * content-drifted files are reported. Files that exist *only* in a mirror are

 * typ

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

#!/usr/bin/env node

/**

 * @fileoverview MCP definition linter CLI.

 * Discovers tool/resource/prompt definitions from conventional locations,

 * runs `validateDefinitions()`, and reports results.

 *

 * Used by devcheck and as a standalone script: `bun run lint:mcp` / `npm run lint:mcp`

 *

 * Discovery strategy:

 *   1. Globs for `*.tool.ts`, `*.resource.ts`, `*.prompt.ts` in known paths

 *   2. Dynamically imports each file

 *   3. Extracts exported definitions by duck-typing (has name/handler/

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

#!/usr/bin/env node

/**

 * @fileoverview MCPB packaging linter — validates env var alignment between

 * `manifest.json` (MCPB bundle install UX) and `server.json` (MCP Registry

 * discovery) for stdio packages.

 *

 * Used by devcheck and as a standalone script: `bun run lint:packaging` /

 * `npm run lint:packaging`.

 *

 * Checks:

 *   1. Manifest `name` must not contain a scope prefix (`@scope/`).

 *   2. Every `user_config` entry must include `title` and `type` fields.

 *   3. Every `${user_con

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a

Evidence

#!/usr/bin/env node

/**

 * @fileoverview One-shot migration: split the monolithic CHANGELOG.md into

 * per-version files under `changelog/<major.minor>.x/<version>.md`.

 *

 * After the initial migration, this script is no longer needed — the

 * per-version files become the source of truth and `scripts/build-changelog.ts`

 * regenerates CHANGELOG.md from them. Keep this around only long enough to

 * verify the round-trip (split → build → diff).

 *

 * Behavior:

 *   • Reads CHANGELOG.md, splits on

Remediation

Replace check-then-use with action-then-handle: Python: `try: with open(p) as f: ... except FileNotFoundError: ...` Node: `try { fs.readFileSync(p); } catch (e) { if (e.code === "ENOENT") ... }` For atomic file creation: Python: `os.open(p, os.O_RDWR | os.O_CREAT | os.O_EXCL)` Node: `fs.open(p, "wx")` — fails if file exists, no race. When you genuinely must check first, use `flock` (Python `fcntl`) or a similar per-process advisory lock to make the window uninteresting to a