# llms-full.txt — lauren-mcp complete API reference

> This file documents every public symbol in `lauren_mcp.__all__`.
> It is intended for AI assistants and LLM-based tooling.
> See llms.txt for a shorter overview.

Package: lauren-mcp
Version: 1.0.0
Status: Beta (Development Status :: 4 - Beta)
Source: https://github.com/lauren-framework/lauren-mcp
Docs: https://lauren-framework.github.io/lauren-mcp/
PyPI: https://pypi.org/project/lauren-mcp/

---

## Version constants

### LATEST

```python
LATEST: str
```

The latest MCP protocol version string supported by this library.
Example value: `"2024-11-05"`.
Use this when constructing `InitializeParams` if you want to negotiate the newest
protocol version.

### STABLE

```python
STABLE: str
```

The stable MCP protocol version string recommended for production use.
May lag behind `LATEST` during transition periods to give server implementations
time to adopt new protocol features.
Example value: `"2024-11-05"`.

### SUPPORTED

```python
SUPPORTED: list[str]
```

A list of all MCP protocol version strings this library can handle.
Example value: `["2024-11-05"]`.
The library refuses `initialize` handshakes that propose a version not in this list.

---

## Server decorators

### mcp_server

```python
def mcp_server(
    path: str,
    *,
    transport: str = "ws",
) -> Callable[[type], type]:
```

Class decorator that registers a class as an MCP server mounted at `path`. Also applies
`@injectable(scope=Scope.SINGLETON)` so the class participates in Lauren's DI graph.

**Parameters**

- `path` (`str`, required): URL path prefix for this server (e.g. `"/mcp"`).
- `transport` (`str`, default `"ws"`): Default transport when `McpServerModule.for_root()`
  is called without an explicit `transport=` argument. One of `"ws"`, `"sse"`,
  `"streamable"`, `"both"` (WS + SSE), or `"all"` (WS + Streamable HTTP).

**Returns**: The decorated class, unmodified except for attached MCP metadata.

**Example**

```python
from lauren_mcp import mcp_server, mcp_tool

@mcp_server("/mcp")
class ShopServer:
    @mcp_tool()
    async def search(self, query: str) -> list[dict]:
        """Search items. Args: query: Search terms."""
        ...
```

---

### mcp_tool

```python
def mcp_tool(
    *,
    name: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    timeout: float | None = None,
    tags: frozenset[str] | set[str] | None = None,
    meta: dict[str, Any] | None = None,
    output_schema: type | dict | None = None,
) -> Callable[[Callable], Callable]:
```

Method decorator that marks an async method on an `@mcp_server` class as an MCP tool.

The tool's JSON Schema is derived automatically from Python type annotations. Parameters
annotated `McpToolContext` are **excluded** from the schema and injected at call time.
Google/Sphinx/NumPy docstring `Args:` sections populate per-parameter descriptions.

**Parameters**

- `name`: Override tool name (defaults to method name).
- `description`: Override tool description (defaults to docstring first paragraph).
- `annotations`: `ToolAnnotations` hints for AI clients (`readOnlyHint`, etc.).
- `timeout`: Per-call execution deadline in seconds; raises `INTERNAL_ERROR` if exceeded.
- `tags`: Categorical tags included in `tools/list` under `"tags"`.
- `meta`: Opaque metadata forwarded to clients under `"_meta"`.
- `output_schema`: JSON Schema dict, Pydantic model, dataclass, or TypedDict describing
  the structured output; advertised as `outputSchema` in `tools/list`.

**Schema generation — supported annotations**

| Python annotation | JSON Schema |
|---|---|
| `str`, `int`, `float`, `bool` | primitive types |
| `list[X]` | `{"type": "array", "items": <X>}` |
| `dict[str, X]` | `{"type": "object", "additionalProperties": <X>}` |
| `X \| None` / `Optional[X]` | optional (omitted from `required`) |
| `Literal["a", "b"]` | `{"type": "string", "enum": ["a", "b"]}` |
| `Annotated[int, Field(ge=0)]` | `{"type": "integer", "minimum": 0}` |
| Pydantic `BaseModel` | object with `$defs` (requires `[pydantic]` extra) |
| `msgspec.Struct` | object with `$defs` (requires `[msgspec]` extra) |
| `@dataclass` class | object schema inlined |
| `TypedDict` | object schema with `required` from `__required_keys__` |
| `UUID`, `datetime`, `date`, `Path` | string with format |
| `McpToolContext` | **excluded** from schema, injected by framework |

**Per-method Lauren decorators**

`@use_guards`, `@use_interceptors`, `@use_exception_handlers`, and `@set_metadata`
from the Lauren framework can be applied to individual `@mcp_tool` methods.
`@mcp_tool()` must be the **outermost** decorator; Lauren decorators go **inside**:

```python
from lauren import use_guards, use_interceptors, use_exception_handlers, set_metadata
from lauren_mcp import mcp_tool

@set_metadata("required_role", "admin")   # readable via ctx.get_metadata(...)
@use_guards(AdminGuard)                   # runs before method; False → FORBIDDEN
@use_interceptors(AuditInterceptor)       # wraps the call; call call_handler.handle()
@use_exception_handlers(DomainErrHandler) # maps domain exceptions to isError: True
@mcp_tool()
async def admin_op(self, ctx: McpToolContext) -> dict: ...
```

See also: `McpExecutionContext` (passed to guards/interceptors), `McpForbiddenError`
(guard rejection), `McpCallHandler` (`call_handler.handle()` in interceptors).

**Example**

```python
from lauren_mcp import mcp_tool, ToolAnnotations, McpToolContext

@mcp_tool(
    annotations=ToolAnnotations(readOnlyHint=True),
    timeout=30.0,
    tags={"search"},
)
async def search(
    self,
    query: str,
    limit: int = 10,
    ctx: McpToolContext,
) -> list[dict]:
    """Search the catalogue.

    Args:
        query: Full-text search query.
        limit: Max results (default 10).
    """
    await ctx.report_progress(0, total=100)
    ...
```

---

### mcp_resource

```python
def mcp_resource(
    uri_template: str,
    *,
    name: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
) -> Callable[[Callable], Callable]:
```

Method decorator that exposes a URI-addressable resource on an `@mcp_server` class.

**URI template operators supported (RFC 6570 subset)**

| Template | Matches | Example |
|---|---|---|
| `{param}` | Single path segment (no `/`) | `/items/{id}` |
| `{+param}` or `{param*}` | Multi-segment (crosses `/`) | `/files/{+path}` |
| `{?p1,p2}` | Optional query parameters | `/search/{q}{?page,size}` |

Path and query variables are passed as keyword arguments; type hints on the method
drive coercion (`int`, `float`, `bool`, `Optional[X]` supported).

**Return types**

| Return type | Wire format |
|---|---|
| `str` | `text` field |
| `bytes` | base64-encoded `blob` field (MIME from `mime_type=`) |
| `BlobResource(data, mime_type)` | blob with explicit MIME |
| `ResourceResult(contents=[...])` | multi-item response |
| `dict` | passed through as-is |

**Example**

```python
@mcp_resource("/files/{+path}{?version}", mime_type="text/plain")
async def read_file(self, path: str, version: int = 1) -> str:
    """Read a versioned file."""
    return load(path, version)

@mcp_resource("/img/{name}", mime_type="image/png")
async def image(self, name: str) -> bytes:
    """Return raw PNG bytes."""
    return open(f"{name}.png", "rb").read()
```

---

### mcp_prompt

```python
def mcp_prompt(
    name: str | None = None,
    *,
    description: str | None = None,
    title: str | None = None,
) -> Callable[[Callable], Callable]:
```

Method decorator that exposes a parameterised prompt template on an `@mcp_server`
class. The decorated method must be async and must return a `str`.

**Parameters**

- `name` (`str | None`, default `None`): Override prompt name. Defaults to method name.
- `description` (`str | None`, default `None`): Override description. Defaults to
  docstring.
- `title` (`str | None`, default `None`): Human-readable display name shown in client UIs.

**Example**

```python
@mcp_prompt()
async def summarise_prompt(self, topic: str = "general") -> str:
    """Build a summarisation prompt.

    Args:
        topic: What to summarise (default "general").
    """
    return f"Please provide a concise summary of the following {topic} content:"
```

---

### mcp_lifespan

```python
# from lauren_mcp.server import mcp_lifespan

@mcp_lifespan          # no parentheses
async def lifespan(self) -> AsyncGenerator[dict, None]:
    ...
```

Method decorator (no arguments) that marks an async generator method as the server's
lifespan hook. The generator runs **once** at server startup; the `dict` it yields
becomes `McpToolContext.lifespan_context` for every subsequent tool call. Code after
the `yield` (typically in a `finally` block) runs at server shutdown.

**Rules**

- Must be an `async def` with a single `yield`.
- Only one `@mcp_lifespan` method per `@mcp_server` class.
- The yielded value must be a `dict` (or `None` for an empty context).

**Example**

```python
from lauren_mcp import mcp_server, mcp_tool, McpToolContext
from lauren_mcp.server import mcp_lifespan

@mcp_server("/mcp")
class MyServer:
    @mcp_lifespan
    async def lifespan(self):
        db = await Database.connect(DATABASE_URL)
        try:
            yield {"db": db}
        finally:
            await db.close()

    @mcp_tool()
    async def query(self, sql: str, ctx: McpToolContext) -> list:
        return await ctx.lifespan_context["db"].fetch(sql)
```

---

### mcp_completion

```python
# from lauren_mcp.server import mcp_completion

def mcp_completion(
    target: str,
    argument: str,
    *,
    ref_type: str = "ref/prompt",
) -> Callable[[Callable], Callable]:
```

Method decorator that registers a completion handler for a prompt argument or resource
template variable. Called by the client when the user types partial input into a prompt
or resource field that supports auto-complete.

**Parameters**

- `target` (`str`): Name of the prompt (for `ref_type="ref/prompt"`) or the URI
  template of the resource (for `ref_type="ref/resource"`).
- `argument` (`str`): The argument name within that prompt or resource for which
  this function provides completions.
- `ref_type` (`str`, default `"ref/prompt"`): Either `"ref/prompt"` or `"ref/resource"`.

The decorated method must be `async def` accepting a single `partial: str` argument
and returning either `list[str]` or `CompletionResult`.

**Example**

```python
from lauren_mcp import mcp_server, mcp_prompt
from lauren_mcp.server import mcp_completion

@mcp_server("/mcp")
class MyServer:
    NAMES = ["Alice", "Bob", "Carol"]

    @mcp_prompt()
    async def greet(self, name: str) -> str:
        """Greet someone. Args: name: Person's name."""
        return f"Hello {name}!"

    @mcp_completion("greet", "name")
    async def complete_greet_name(self, partial: str) -> list[str]:
        return [n for n in self.NAMES if n.lower().startswith(partial.lower())]
```

---

### McpServerModule

```python
class McpServerModule:
    @staticmethod
    def for_root(
        server_cls: type,
        *,
        transport: str = "ws",
        server_info: Implementation | None = None,
        capabilities: ServerCapabilities | None = None,
        providers: list | None = None,
        imports: list | None = None,
        exports: list | None = None,
        log_level: str = "debug",
        mounts: list[tuple[type, str]] | None = None,
        proxies: list[tuple[McpClientProtocol, str]] | None = None,
    ) -> type:  # returns a Lauren @module class
        ...
```

Builds a Lauren `@module` that mounts *server_cls* in the DI graph and
wires all MCP handler coroutines.

**Parameters**

- `server_cls`: Class decorated with `@mcp_server`.
- `transport`: One of `"ws"`, `"sse"`, `"streamable"`, `"both"` (WS+SSE), `"all"` (WS+Streamable).
- `server_info`: Override the name/version in the `initialize` response.
- `capabilities`: Override auto-detected `ServerCapabilities`.
- `providers`: Extra Lauren DI providers.
- `imports`: Extra Lauren `@module` classes to import.
- `exports`: Extra types to export from the generated module.
- `log_level`: Minimum severity for `ctx.log()` notifications (`"debug"` / `"info"` / `"warning"` / `"error"`).
- `mounts`: `[(OtherServerCls, "prefix_"), ...]` — expose another `@mcp_server` class's tools/resources/prompts with name prefix.
- `proxies`: `[(client, "prefix_"), ...]` — connect a remote MCP client at startup and re-export its tools locally.

**Raises** `TypeError` if *server_cls* is not decorated with `@mcp_server`.

**Route mounting** for `@mcp_server("/mcp")`:

| Transport value | Endpoints |
|---|---|
| `"ws"` | `ws://host/mcp/ws` |
| `"sse"` | `GET /mcp/sse` (stream) + `POST /mcp/` (messages) |
| `"streamable"` | `POST /mcp/` + `GET /mcp/` (push) + `DELETE /mcp/` |
| `"both"` | WS + SSE |
| `"all"` | WS + Streamable HTTP |

**Example**

```python
from lauren import LaurenFactory, module
from lauren_mcp import McpServerModule, McpServer

@module(imports=[McpServerModule.for_root(
    MyServer,
    transport="all",
    log_level="info",
    mounts=[(AnalyticsServer, "analytics_")],
    proxies=[(McpServer.streamable_http("http://remote/mcp"), "remote_")],
)])
class AppModule:
    pass

app = LaurenFactory.create(AppModule)
```

---

### McpToolContext

```python
@dataclass(frozen=True)
class McpToolContext:
    tool_name: str
    tool_use_id: str | int | None
    headers: Headers | None          # lauren.types.Headers (case-insensitive)
    execution_context: ExecutionContext | None  # lauren.types.ExecutionContext
    session_id: str | None
    metadata: dict[str, Any]         # from @set_metadata on the @mcp_server class
    state: dict[str, Any]            # mutable per-call scratch space
    extras: dict[str, Any]           # extension bag
    lifespan_context: dict[str, Any] # dict yielded by @mcp_lifespan
```

Context injected into `@mcp_tool` methods when a parameter is annotated
`McpToolContext`. The parameter may have any name; it is **excluded** from the JSON
schema so AI clients never see or supply it.

Declare it anywhere in the signature:

```python
@mcp_tool()
async def my_tool(self, query: str, ctx: McpToolContext) -> str: ...

@mcp_tool()
async def other(self, data: str, tool_ctx: McpToolContext | None = None) -> str: ...
```

**Methods**

```python
ctx.get_metadata(key, default=None)   # shorthand for ctx.metadata.get(...)

# Progress (sends notifications/progress — no-op when no progressToken)
await ctx.report_progress(progress: float | int, total: float | int | None = None)

# Structured log to client (sends notifications/message)
await ctx.log(level: Literal["debug","info","warning","error"], message: str, data: dict | None = None)
await ctx.debug(message, data=None)
await ctx.info(message, data=None)
await ctx.warning(message, data=None)
await ctx.error(message, data=None)

# Server-initiated LLM call (WS and Streamable HTTP only)
result: CreateMessageResult = await ctx.sample(
    messages: str | list[SamplingMessage],
    *,
    max_tokens: int = 1024,
    system_prompt: str | None = None,
    temperature: float | None = None,
    result_type: type | None = None,   # parse + validate JSON reply
)

# Server asks client to prompt user (WS and Streamable HTTP only)
answer: ElicitResult = await ctx.elicit(
    message: str,
    response_type = None,  # None | str | bool | int | Literal | dataclass | TypedDict | BaseModel
)
```

**Raises**

- `McpSamplingNotAvailable` — `ctx.sample()` called on legacy SSE transport, or client
  did not advertise the `sampling` capability.
- `McpElicitationNotAvailable` — same for `ctx.elicit()`.

---

### ToolAnnotations

```python
@dataclass(frozen=True)
class ToolAnnotations:
    readOnlyHint: bool = False    # tool does not modify state
    destructiveHint: bool = True  # changes cannot be undone (conservative default)
    idempotentHint: bool = False  # repeated calls produce same result
    openWorldHint: bool = True    # tool may interact with external systems (conservative)
```

Behavioural hints transmitted to AI clients in `tools/list`. Clients use them to
decide confirmation UX (e.g. ask before calling a destructive tool). The defaults
follow the MCP specification's conservative assumptions.

```python
from lauren_mcp import mcp_tool, ToolAnnotations

@mcp_tool(annotations=ToolAnnotations(readOnlyHint=True, destructiveHint=False))
async def search(self, q: str) -> list: ...
```

---

## Client

### McpServer

```python
class McpServer:
    @staticmethod
    def stdio(
        command: list[str] | tuple[str, ...],
        *,
        max_retries: int = 3,
        startup_timeout: float = 10.0,
    ) -> McpClientProtocol: ...

    @staticmethod
    def ws(
        url: str,
        *,
        headers: dict[str, str] | None = None,
        max_retries: int = 3,
        startup_timeout: float = 10.0,
    ) -> McpClientProtocol: ...

    @staticmethod
    def http(
        url: str,
        *,
        headers: dict[str, str] | None = None,
        max_retries: int = 3,
        startup_timeout: float = 10.0,
        **feature_kwargs,
    ) -> McpClientProtocol: ...

    @staticmethod
    def streamable_http(
        url: str,
        *,
        headers: dict[str, str] | None = None,
        max_retries: int = 3,
        startup_timeout: float = 10.0,
        **feature_kwargs,
    ) -> McpClientProtocol: ...
```

Factory class for creating MCP client connections. Never instantiate directly.

All four factories accept these additional **feature kwargs**:

| Kwarg | Type | Description |
|---|---|---|
| `protocol_version` | `str` | Protocol version to advertise (default `"2025-03-26"`) |
| `roots` | `list[Root]` or `Callable` | Filesystem roots exposed to the server |
| `progress_handler` | `Callable` | Invoked on `notifications/progress` |
| `log_handler` | `Callable` | Invoked on `notifications/message` |
| `list_changed_handler` | `Callable` | Invoked on tool/resource/prompt `list_changed` |
| `sampling_handler` | `async Callable` | Responds to `sampling/createMessage` requests |
| `elicitation_handler` | `async Callable` | Responds to `elicitation/create` requests |

**`McpServer.stdio`** — subprocess stdin/stdout transport. No extra deps required.

**`McpServer.ws`** — WebSocket (MCP 2025-03-26). Requires `pip install "lauren-mcp[ws]"`.

**`McpServer.http`** — Legacy HTTP+SSE (MCP 2024-11-05). Requires `pip install "lauren-mcp[http]"`.

**`McpServer.streamable_http`** — Streamable HTTP (MCP 2025-03-26, recommended).
Requires `pip install "lauren-mcp[http]"`.

**Example**

```python
from lauren_mcp import McpServer, Root

stdio_client = McpServer.stdio(["python", "-m", "my_server"])
ws_client    = McpServer.ws("wss://api.example.com/mcp/ws",
                             headers={"Authorization": "Bearer token"},
                             progress_handler=lambda p: print(p["progress"]))
http_client  = McpServer.streamable_http("https://api.example.com/mcp",
                                          roots=[Root("file:///workspace")])
```

---

### McpClientProtocol

```python
class McpClientProtocol(Protocol):
    # Lifecycle
    async def connect(self) -> None: ...
    async def close(self) -> None: ...
    # Tools
    async def list_tools(self) -> list[ToolSchema]: ...
    async def call_tool(self, name: str, arguments: dict | None = None) -> Any: ...
    # Resources
    async def list_resources(self) -> list[ResourceSchema]: ...
    async def read_resource(self, uri: str) -> Any: ...
    # Prompts
    async def list_prompts(self) -> list[PromptSchema]: ...
    async def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> Any: ...
    # Utility
    async def ping(self) -> None: ...
    # Protocol version (available after connect())
    @property
    def protocol_version(self) -> str: ...
    # Notification subscriptions
    def on_progress(self, handler: Callable) -> Callable[[], None]: ...
    def on_log(self, handler: Callable) -> Callable[[], None]: ...
    def on_list_changed(self, handler: Callable) -> Callable[[], None]: ...
    # Roots
    async def notify_roots_changed(self) -> None: ...
```

The interface implemented by all four transport clients. You receive instances of this
from `McpServer.stdio/ws/http/streamable_http`.

**Lifecycle methods**

- `connect()`: Establishes the transport and runs the MCP `initialize` handshake.
- `close()`: Graceful teardown.

**Protocol version**

- `protocol_version` (property): The version negotiated with the server. Only available
  after `connect()`.

**Notification subscriptions** — each returns an unsubscribe callable:

- `on_progress(handler)`: Called with `{"progressToken": ..., "progress": ..., "total": ...}`.
- `on_log(handler)`: Called with `{"level": "info", "logger": "...", "data": {...}}`.
- `on_list_changed(handler)`: Called with `"tools"`, `"resources"`, or `"prompts"`.

**Roots**

- `notify_roots_changed()`: Send `notifications/roots/list_changed` to the server.
  Only valid when `roots=` was supplied to the factory.

**Data methods**

- `list_tools()` → `list[ToolSchema]`
- `call_tool(name, arguments)` → raw dict `{"content": [...], "isError": False}`;
  raises `McpCallError` on server error
- `list_resources()` → `list[ResourceSchema]`
- `read_resource(uri)` → raw dict `{"contents": [...]}`
- `list_prompts()` → `list[PromptSchema]`
- `get_prompt(name, arguments)` → raw dict `{"messages": [...], "description": ...}`
- `ping()` → `None`; raises `McpCallError` on failure

**Example**

```python
import asyncio, json
from lauren_mcp import McpServer

async def main():
    client = McpServer.stdio(["python", "-m", "my_server"])
    await client.connect()
    tools = await client.list_tools()
    result = await client.call_tool("search", {"query": "coffee"})
    print(json.loads(result["content"][0]["text"]))
    await client.close()

asyncio.run(main())
```

---

### McpCallError

```python
class McpCallError(Exception):
    code: int   # JSON-RPC error code, e.g. -32603 (INTERNAL_ERROR)
```

Raised by `call_tool()`, `list_tools()`, and other client methods when the server
returns a JSON-RPC error response.

```python
from lauren_mcp import McpCallError

try:
    result = await client.call_tool("divide", {"a": 1, "b": 0})
except McpCallError as exc:
    print(f"Server error {exc.code}: {exc}")
```

Common codes: `PARSE_ERROR = -32700`, `METHOD_NOT_FOUND = -32601`,
`INTERNAL_ERROR = -32603`.

---

### McpServerConfig

```python
from dataclasses import dataclass

@dataclass
class McpServerConfig:
    alias: str
    client: Any   # McpClientProtocol
```

Configuration dataclass pairing a short alias with an `McpClientProtocol` instance.
Used with `McpToolBridge` and `lauren_ai.AgentModule.for_root(mcp_servers=[...])`.

**Fields**

- `alias` (`str`, required): Short identifier. Tools from this server are namespaced as
  `alias__tool_name`.
- `client` (required): Client instance from `McpServer.stdio/ws/http`.

**Example**

```python
from lauren_mcp import McpServerConfig, McpServer

config = McpServerConfig(
    alias="fs",
    client=McpServer.stdio(["python", "-m", "my_mcp_server"]),
)
```

---

### McpToolBridge

```python
class McpToolBridge:
    def __init__(self, servers: list[McpServerConfig]) -> None: ...
    def set_registry(self, registry: Any) -> None: ...
    async def connect_all(self) -> None: ...
    async def disconnect_all(self) -> None: ...
```

Manages the lifecycle for a list of `McpServerConfig` entries — connects every
server, optionally populates a registry, and disconnects cleanly.

Used internally by `lauren_ai.AgentModule` when `mcp_servers=[...]` is passed.

**Methods**

- `set_registry(registry)`: Attach any object with a `register_mcp_server(alias,
  tools, client)` method.  Called by `connect_all()` once per server after tools are
  fetched.
- `connect_all()`: Connect all servers and call `registry.register_mcp_server` for
  each one.  Failures are logged at ERROR and do not abort other servers.
- `disconnect_all()`: Close all clients; individual failures are suppressed.

**Example**

```python
from lauren_mcp import McpToolBridge, McpServerConfig, McpServer

bridge = McpToolBridge([
    McpServerConfig(alias="alpha", client=McpServer.stdio(["python", "server_a.py"])),
    McpServerConfig(alias="beta",  client=McpServer.stdio(["python", "server_b.py"])),
])
await bridge.connect_all()
# ... use connected clients ...
await bridge.disconnect_all()
```

**Methods**

- `get_tool_names()`: Returns the list of namespaced tool names available from this
  server (e.g. `["fs__read_file", "fs__list_directory"]`). Must be called after
  entering the async context manager.
- `get_tool_schemas()`: Returns the raw `ToolSchema` objects for all exposed tools
  (after applying `tool_filter`).
- `call(namespaced_name, arguments)`: Strips the alias prefix and calls the underlying
  tool. Raises `McpToolNotFoundError` if the name is not in this bridge's tool list.

**Example**

```python
from lauren_mcp import McpToolBridge, McpServerConfig, McpServer

config = McpServerConfig(alias="svc", client=McpServer.stdio([...]))
bridge = McpToolBridge(config)

async with bridge:
    names = bridge.get_tool_names()    # ["svc__search", "svc__add"]
    result = await bridge.call("svc__search", {"query": "widget"})
```

---

## Wire types

### JsonRpcRequest

```python
@dataclass
class JsonRpcRequest:
    jsonrpc: str           # always "2.0"
    id: int | str
    method: str
    params: dict | None = None
```

An outgoing or incoming JSON-RPC 2.0 request. Always has an `id`; for fire-and-forget
messages use `JsonRpcNotification`.

---

### JsonRpcNotification

```python
@dataclass
class JsonRpcNotification:
    jsonrpc: str           # always "2.0"
    method: str
    params: dict | None = None
```

A JSON-RPC 2.0 notification (no `id`). The server does not send a response.
Used for MCP lifecycle events like `notifications/initialized`.

---

### JsonRpcResponse

```python
@dataclass
class JsonRpcResponse:
    jsonrpc: str           # always "2.0"
    id: int | str
    result: dict | list | str | int | float | bool | None
```

A successful JSON-RPC 2.0 response. The `result` field contains the method-specific
return value.

---

### JsonRpcErrorResponse

```python
@dataclass
class JsonRpcErrorResponse:
    jsonrpc: str           # always "2.0"
    id: int | str | None
    error: JsonRpcError

@dataclass
class JsonRpcError:
    code: int
    message: str
    data: dict | None = None
```

An error JSON-RPC 2.0 response. `id` is `None` when the error occurred before the
request `id` could be determined (e.g. parse error).

---

### JsonRpcError

```python
@dataclass
class JsonRpcError:
    code: int
    message: str
    data: Any = None
```

The error object embedded inside a `JsonRpcErrorResponse`. `code` is a standard JSON-RPC
or MCP error code (see `McpErrorCode`). `data` carries optional extra detail.

---

### McpErrorCode

```python
from enum import IntEnum

class McpErrorCode(IntEnum):
    PARSE_ERROR        = -32700
    INVALID_REQUEST    = -32600
    METHOD_NOT_FOUND   = -32601
    INVALID_PARAMS     = -32602
    INTERNAL_ERROR     = -32603
    TOOL_NOT_FOUND     = -32001
    RESOURCE_NOT_FOUND = -32002
    PROMPT_NOT_FOUND   = -32003
    CONNECTION_ERROR   = -32004
```

Standard JSON-RPC 2.0 error codes plus MCP-specific extensions.

---

### parse_message

```python
def parse_message(
    data: str | bytes,
) -> JsonRpcRequest | JsonRpcNotification | JsonRpcResponse | JsonRpcErrorResponse:
```

Parse a raw JSON string or bytes into the appropriate JSON-RPC message type.

**Parameters**

- `data` (`str | bytes`): Raw JSON-RPC message from the wire.

**Returns**: One of the four JSON-RPC message dataclasses.

**Raises**: `JsonRpcParseError` if `data` is not valid JSON or does not conform to
JSON-RPC 2.0 structure.

**Example**

```python
from lauren_mcp import parse_message

msg = parse_message('{"jsonrpc":"2.0","id":1,"method":"tools/list","params":null}')
# → JsonRpcRequest(jsonrpc='2.0', id=1, method='tools/list', params=None)
```

---

### build_error_response

```python
def build_error_response(
    request_id: int | str | None,
    code: McpErrorCode | int,
    message: str,
    data: dict | None = None,
) -> JsonRpcErrorResponse:
```

Convenience constructor for `JsonRpcErrorResponse`.

**Parameters**

- `request_id` (`int | str | None`): The `id` from the request that triggered the
  error. Pass `None` for parse errors.
- `code` (`McpErrorCode | int`): Error code (use `McpErrorCode` enum values).
- `message` (`str`): Human-readable error description.
- `data` (`dict | None`, default `None`): Optional additional error detail.

**Example**

```python
from lauren_mcp import build_error_response, McpErrorCode

resp = build_error_response(
    request_id=42,
    code=McpErrorCode.TOOL_NOT_FOUND,
    message="Tool 'foo' is not registered.",
)
```

---

### ToolSchema

```python
@dataclass
class ToolSchema:
    name: str
    description: str
    inputSchema: dict   # JSON Schema object describing the tool's parameters
```

Descriptor for a single MCP tool as returned by `tools/list`.

`inputSchema` is a JSON Schema object with `type: "object"`, `properties`, and
`required` fields. Generated automatically by `@mcp_tool` from Python annotations.

---

### ResourceSchema

```python
@dataclass
class ResourceSchema:
    uri: str
    name: str
    description: str | None = None
    mimeType: str = "text/plain"
```

Descriptor for a single MCP resource as returned by `resources/list`.

---

### Role

```python
Role = Literal["user", "assistant"]
```

Type alias for MCP audience roles. Used in `ResourceAnnotations.audience` and
`SamplingMessage.role`.

---

### ResourceAnnotations

```python
@dataclass
class ResourceAnnotations:
    audience: list[Role] | None = None  # intended readers; None means unrestricted
    priority: float | None = None       # relevance weight in [0.0, 1.0]
```

Annotations attached to an MCP resource for UI and routing hints. Pass to
`@mcp_resource(annotations=...)`.

- `audience`: Restricts which roles should see the resource (`"user"`, `"assistant"`).
- `priority`: Relevance weight in `[0.0, 1.0]`; higher means higher priority.
  Raises `ValueError` if outside the valid range.

```python
from lauren_mcp import ResourceAnnotations
from lauren_mcp.server import mcp_resource

@mcp_resource("/reports/{name}",
    annotations=ResourceAnnotations(audience=["user"], priority=0.8))
async def report(self, name: str) -> str: ...
```

---

### ResourceContent

```python
@dataclass
class ResourceContent:
    uri: str
    mimeType: str | None = None
    text: str | None = None
    blob: str | None = None   # base-64 encoded binary
```

The content of a single resource returned inside a `resources/read` response.
Exactly one of `text` or `blob` is populated.  `text` carries plain-text content;
`blob` carries arbitrary binary data encoded as base-64.

---

### PromptSchema

```python
@dataclass
class PromptSchema:
    name: str
    description: str | None = None
    arguments: list[PromptArgument] = field(default_factory=list)
```

Descriptor for a single MCP prompt as returned by `prompts/list`.

---

### TextContent

```python
@dataclass
class TextContent:
    type: Literal["text"]   # always "text"
    text: str
```

A plain-text content block. The most common return type from tool calls.

**Example**

```python
result = await client.call_tool("search", {"query": "coffee"})
# result[0] is typically TextContent
print(result[0].text)
```

---

### ImageContent

```python
@dataclass
class ImageContent:
    type: Literal["image"]   # always "image"
    data: str                # base-64 encoded image bytes
    mimeType: str            # e.g. "image/png", "image/jpeg"
```

A base-64 encoded image content block. Returned by tools that produce images.

---

### EmbeddedResource

```python
@dataclass
class EmbeddedResource:
    type: Literal["resource"]   # always "resource"
    resource: TextResourceContents | BlobResourceContents
```

An embedded resource returned inside a tool call result. `TextResourceContents` has
`uri: str` and `text: str`; `BlobResourceContents` has `uri: str` and `blob: str`
(base-64 encoded).

---

### AudioContent

```python
@dataclass
class AudioContent:
    data: str      # base64-encoded audio bytes
    mimeType: str  # e.g. "audio/wav", "audio/mpeg", "audio/ogg"
    type: str = "audio"

    @classmethod
    def from_bytes(cls, data: bytes, mime_type: str = "audio/wav") -> AudioContent: ...
```

A base64-encoded audio content block. Returned by tools that produce audio output.
Use `AudioContent.from_bytes(raw_audio, "audio/wav")` to construct from raw bytes
without manually base64-encoding.

---

### ResourceLink

```python
@dataclass
class ResourceLink:
    uri: str
    name: str | None = None
    description: str | None = None
    mimeType: str | None = None
    type: str = "resource_link"
```

A lightweight reference to a resource by URI. Unlike `EmbeddedResource`, a `ResourceLink`
does **not** inline the resource's content — it carries only the URI and optional metadata.
The MCP client may fetch the resource separately via `resources/read` if needed.

---

### AnyContent

```python
AnyContent = TextContent | ImageContent | AudioContent | EmbeddedResource | ResourceLink
```

Union alias for any content item that can appear in a tool result or message.
Use this as a type annotation when a value may be any of the five content types.

---

### PromptArgument

```python
@dataclass
class PromptArgument:
    name: str
    description: str | None = None
    required: bool = False
```

A single argument descriptor within a `PromptSchema`. Generated automatically from
`@mcp_prompt` method parameters.

---

### PromptMessage

```python
@dataclass
class PromptMessage:
    role: Literal["user", "assistant"]
    content: TextContent | ImageContent | EmbeddedResource
```

A single rendered message within a `GetPromptResult`.

---

### InitializeParams

```python
@dataclass
class InitializeParams:
    protocolVersion: str
    capabilities: ClientCapabilities
    clientInfo: Implementation
```

Payload sent by the client in the MCP `initialize` request. Typically constructed
internally by the client transport; you rarely need to create this directly.

---

### InitializeResult

```python
@dataclass
class InitializeResult:
    protocolVersion: str
    capabilities: ServerCapabilities
    serverInfo: Implementation
    instructions: str | None = None
```

Payload returned by the server in the `initialize` response. Available on the client
after `connect()` as `client.server_info`.

---

### ClientCapabilities

```python
@dataclass
class ClientCapabilities:
    roots: dict | None = None
    sampling: dict | None = None
    elicitation: dict | None = None
    experimental: dict | None = None
```

Capability flags advertised by the client during the MCP handshake. Non-`None` values
signal support: `roots` for filesystem roots, `sampling` for server-initiated LLM calls,
`elicitation` for server-initiated user prompts.

---

### ServerCapabilities

```python
@dataclass
class ServerCapabilities:
    tools: dict | None = None
    resources: dict | None = None
    prompts: dict | None = None
    logging: dict | None = None
    experimental: dict | None = None
```

Capability flags advertised by the server during the MCP handshake.
The presence of `tools`, `resources`, or `prompts` keys indicates support for the
corresponding MCP feature.

---

### Implementation

```python
@dataclass
class Implementation:
    name: str
    version: str
```

Identifies a client or server implementation in the MCP handshake messages.
For `lauren-mcp` servers the default is `Implementation(name="lauren-mcp", version=<package version>)`.

---

### ToolCallParams

```python
@dataclass
class ToolCallParams:
    name: str
    arguments: dict | None = None
```

The `params` payload for a `tools/call` JSON-RPC request.

---

### ToolResult

```python
@dataclass
class ToolResult:
    content: list[TextContent | ImageContent | EmbeddedResource]
    isError: bool = False
    structuredContent: dict[str, Any] | None = None
```

Full result envelope from a `tools/call` response. `structuredContent` carries a typed
JSON object alongside the human-readable `content` blocks (MCP 2025-06-18 structured
output). When `isError=True`, the blocks describe the error.

---

### ReadResourceParams

```python
@dataclass
class ReadResourceParams:
    uri: str
```

The `params` payload for a `resources/read` JSON-RPC request.

---

### ReadResourceResult

```python
@dataclass
class ReadResourceResult:
    contents: list[TextResourceContents | BlobResourceContents]
```

Result envelope from a `resources/read` response.

---

### GetPromptParams

```python
@dataclass
class GetPromptParams:
    name: str
    arguments: dict[str, str] | None = None
```

The `params` payload for a `prompts/get` JSON-RPC request.

---

### GetPromptResult

```python
@dataclass
class GetPromptResult:
    description: str | None
    messages: list[PromptMessage]
```

Result envelope from a `prompts/get` response. `messages` contains the rendered prompt
as a list of role-tagged content blocks, ready to prepend to an LLM conversation.

---

### McpParseError

```python
class McpParseError(ValueError):
    ...
```

Raised by `parse_message()` when a raw JSON string cannot be decoded or does not
conform to JSON-RPC 2.0 structure (missing `jsonrpc` field, unknown message shape, etc.).
It is a subclass of `ValueError`.

---

### ToolOutput

```python
@dataclass
class ToolOutput:
    content: list[Any] | None = None
    structured_content: dict[str, Any] | None = None
    is_error: bool = False
```

Rich return type for `@mcp_tool` methods. Lets a tool control the human-readable
`content` blocks and the machine-readable `structured_content` dict independently.
When `content` is `None` and `structured_content` is set, a JSON text block is
auto-generated.

```python
from lauren_mcp import mcp_tool, ToolOutput, TextContent

@mcp_tool()
async def analyse(self, text: str) -> ToolOutput:
    return ToolOutput(
        content=[TextContent(text="Sentiment: positive")],
        structured_content={"sentiment": "positive", "score": 0.92},
    )
```

---

### BlobResource

```python
@dataclass
class BlobResource:
    data: bytes
    mime_type: str = "application/octet-stream"
```

Explicit blob return type for `@mcp_resource` methods. The `data` bytes are
base64-encoded into `ResourceContent.blob`. Using `BlobResource` lets you override the
MIME type independently of the `mime_type=` parameter on `@mcp_resource`.

```python
from lauren_mcp import BlobResource

@mcp_resource("/reports/{name}")
async def report(self, name: str) -> BlobResource:
    return BlobResource(data=generate_pdf(name), mime_type="application/pdf")
```

---

### ResourceResult

```python
@dataclass
class ResourceResult:
    contents: list[Any]
```

Multi-item return type for `@mcp_resource` methods. Each item in `contents` may be
a `str`, `bytes`, `BlobResource`, `ResourceContent`, or a dict.

```python
from lauren_mcp import ResourceResult

@mcp_resource("/bundle/{id}")
async def bundle(self, id: str) -> ResourceResult:
    return ResourceResult(contents=["readme text", b"\x89PNG..."])
```

---

### ToolUseContent

```python
@dataclass
class ToolUseContent:
    id: str                 # opaque identifier for this tool call
    name: str               # tool name
    input: dict[str, Any]   # tool arguments
    type: str = "tool_use"

    def to_dict(self) -> dict[str, Any]: ...

    @classmethod
    def from_dict(cls, obj: dict[str, Any]) -> ToolUseContent: ...
```

A tool invocation block inside a `SamplingMessage`. Represents the LLM requesting
to call a tool — returned in a `sampling/createMessage` response when the model
decides to use a tool, or included when re-sending past assistant turns that contained
tool calls. The `id` must match the `tool_use_id` of the corresponding `ToolResultContent`.

---

### ToolResultContent

```python
@dataclass
class ToolResultContent:
    tool_use_id: str                          # must match ToolUseContent.id
    content: list[TextContent | ImageContent] = field(default_factory=list)
    is_error: bool = False
    type: str = "tool_result"

    def to_dict(self) -> dict[str, Any]: ...

    @classmethod
    def from_dict(cls, obj: dict[str, Any]) -> ToolResultContent: ...
```

The result of a prior tool invocation, placed in a `role="user"` `SamplingMessage`
to continue the conversation after a `ToolUseContent`. `is_error=True` signals that
the tool call failed; the LLM can decide how to proceed.

---

### SamplingMessage

```python
@dataclass
class SamplingMessage:
    role: Literal["user", "assistant"]
    content: TextContent | ImageContent
```

A single message in a `sampling/createMessage` request sent via `ctx.sample()`.

---

### CreateMessageParams

```python
@dataclass
class CreateMessageParams:
    messages: list[SamplingMessage]
    maxTokens: int
    systemPrompt: str | None = None
    includeContext: Literal["none", "thisServer", "allServers"] = "none"
    temperature: float | None = None
    stopSequences: list[str] = field(default_factory=list)
    modelPreferences: dict[str, Any] | None = None
    metadata: dict[str, Any] | None = None
```

Parameters for a `sampling/createMessage` server-to-client request. Built automatically
when you call `ctx.sample()`.

---

### CreateMessageResult

```python
@dataclass
class CreateMessageResult:
    role: Literal["assistant"]
    content: TextContent | ImageContent
    model: str
    stopReason: str | None = None

    @property
    def text(self) -> str: ...   # shorthand for content.text
```

Result of a `sampling/createMessage` request. Returned by `ctx.sample()`.

---

### validate_sampling_messages

```python
def validate_sampling_messages(messages: list[SamplingMessage]) -> None:
```

Validate that `ToolResultContent` blocks are properly paired with preceding
`ToolUseContent` blocks in a sampling message list.

**Raises** `ValueError` if a `ToolResultContent` appears before a `ToolUseContent` with
the matching `id`, or if the ordering is otherwise invalid. An empty list is valid.
This performs a shallow ordering check only — it does not validate tool input schemas
or result content types.

```python
from lauren_mcp import validate_sampling_messages, SamplingMessage, TextContent

msgs = [SamplingMessage(role="user", content=TextContent(text="Hello"))]
validate_sampling_messages(msgs)  # passes — no tool use blocks
```

---

### ElicitResult

```python
@dataclass
class ElicitResult:
    action: Literal["accept", "decline", "cancel"]
    content: dict[str, Any] | None = None
```

Result of an `elicitation/create` server-to-client request. Returned by `ctx.elicit()`.
`content` is populated when `action == "accept"`.

```python
answer = await ctx.elicit("Delete 500 records?")
if answer.action == "accept":
    ...
```

---

### UrlElicitResult

```python
@dataclass
class UrlElicitResult:
    action: Literal["accept", "cancel"]
```

Result of a URL-elicitation `elicitation/create` request. Unlike `ElicitResult`, there
is no `"decline"` state — the only outcomes are completion (`"accept"`) or abandonment
(`"cancel"`). No `content` field is present; the URL flow itself carries any data.

---

### McpUrlElicitationNotAvailable

```python
class McpUrlElicitationNotAvailable(RuntimeError): ...
```

Raised by `ctx.elicit_url()` when the connected client did not advertise the
`urlElicitation` sub-capability, the `elicitation` capability is absent, or the
transport cannot carry server-to-client requests.

---

### Root

```python
@dataclass
class Root:
    uri: str            # file:// URI
    name: str | None = None
```

A filesystem root exposed by an MCP client to servers that request them via
`roots/list`. Pass a list to `McpServer.*(..., roots=[Root("file:///workspace")])`.

---

### CompletionResult

```python
@dataclass
class CompletionResult:
    values: list[str]
    total: int | None = None
    has_more: bool = False
```

Result of a `completion/complete` request. Returned by `@mcp_completion` handlers
(or by the framework after collecting them). `values` contains the completion strings
to offer to the user. `total` reports the total number of matches when only a subset
is returned; `has_more` signals there are additional results beyond `values`.

```python
from lauren_mcp import CompletionResult
from lauren_mcp.server import mcp_completion

@mcp_completion("greet", "name")
async def complete_name(self, partial: str) -> CompletionResult:
    matches = [n for n in self.NAMES if n.lower().startswith(partial.lower())]
    return CompletionResult(values=matches[:5], total=len(matches), has_more=len(matches) > 5)
```

---

### McpSamplingNotAvailable

```python
class McpSamplingNotAvailable(RuntimeError): ...
```

Raised by `ctx.sample()` when:
- The transport is legacy HTTP+SSE (server-to-client requests not supported).
- The connected client did not advertise the `sampling` capability.

---

### McpElicitationNotAvailable

```python
class McpElicitationNotAvailable(RuntimeError): ...
```

Raised by `ctx.elicit()` under the same conditions as `McpSamplingNotAvailable`.

---

### McpToolNameCollision

```python
class McpToolNameCollision(Exception): ...
```

Raised at server startup when two composition sources (from `mounts=` or `proxies=`)
produce the same tool name after applying the prefix. Fix by choosing a different
prefix or renaming the conflicting tool.

---

### make_mount_binder

```python
# from lauren_mcp.server import make_mount_binder

def make_mount_binder(mounted_cls: type, prefix: str) -> type:
```

Build an `@injectable(Singleton)` class that registers all tools, resources, and prompts
from `mounted_cls` into the shared catalogue with `prefix` applied to their names.
Add the returned class (and `mounted_cls` itself) to `McpServerModule.for_root(providers=[...])`.

This is the low-level primitive behind `McpServerModule.for_root(mounts=[...])`. Use
`mounts=` directly unless you need custom DI wiring.

**Raises** `TypeError` if `mounted_cls` is not decorated with `@mcp_server`.

```python
from lauren_mcp.server import make_mount_binder

binder = make_mount_binder(AnalyticsServer, "analytics_")
# pass binder + AnalyticsServer to providers=[...]
```

---

### make_proxy_binder

```python
# from lauren_mcp.server import make_proxy_binder

def make_proxy_binder(client: McpClientProtocol, prefix: str) -> type:
```

Build an `@injectable(Singleton)` class that connects `client` at startup, fetches its
remote tool catalogue, and registers each tool locally under `{prefix}{name}`. At
shutdown it deregisters the tools and closes the connection.

This is the low-level primitive behind `McpServerModule.for_root(proxies=[...])`. Use
`proxies=` directly unless you need custom DI wiring.

```python
from lauren_mcp import McpServer
from lauren_mcp.server import make_proxy_binder

binder = make_proxy_binder(McpServer.streamable_http("http://remote/mcp"), "remote_")
# pass binder to providers=[...]
```

---

### RouteEntry

```python
@dataclass
class RouteEntry:
    pattern: str                                   # regex matched against the path
    method: str | None = None                      # "GET", "POST", … or None for all
    expose_as: Literal["tool", "exclude"] = "tool"
    name_override: str | None = None
    description_override: str | None = None
```

One rule controlling how an OpenAPI operation maps to an MCP tool.
Pass a list of `RouteEntry` objects as `route_map=` to `build_openapi_server_class()`.
The first matching rule wins; operations with no match default to `"tool"`.

---

### build_openapi_server_class

```python
# from lauren_mcp.server import build_openapi_server_class

def build_openapi_server_class(
    spec: dict | str | Path,
    *,
    http_client: Any,              # httpx.AsyncClient or compatible
    base_url: str = "",
    server_path: str = "/mcp",
    route_map: list[RouteEntry] | None = None,
    class_name: str = "OpenApiMcpServer",
) -> type:
```

Generates an `@mcp_server` class whose tools call a REST API described by an OpenAPI
3.x spec. Pass the result to `McpServerModule.for_root()`.

**Parameters**

- `spec`: Parsed dict, or path to a `.json` / `.yaml` file.
- `http_client`: `httpx.AsyncClient` used to execute requests.
- `base_url`: URL prefix (may be empty when the client has a `base_url`).
- `server_path`: Mount path passed to `@mcp_server`.
- `route_map`: Ordered `RouteEntry` rules; first match wins.
- `class_name`: Name for the generated class (for tracebacks).

> **Caveat**: Auto-generated tool descriptions are lower quality than hand-crafted
> ones. Recommended for prototyping, not production.

```python
import httpx, json
from lauren_mcp.server import build_openapi_server_class, RouteEntry

spec = json.load(open("openapi.json"))
ApiServer = build_openapi_server_class(
    spec,
    http_client=httpx.AsyncClient(base_url="https://api.example.com"),
    route_map=[RouteEntry(r"/admin", expose_as="exclude")],
)
```

---

## Built-in resource classes

### FileResource

```python
# from lauren_mcp.server import FileResource

class FileResource:
    def __init__(
        self,
        path: str | Path,
        uri: str,
        *,
        name: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
    ) -> None: ...
```

Expose a local file as a static MCP resource. MIME type is auto-detected from the file
extension when `mime_type` is omitted. Text-like MIME types (`text/*`,
`application/json`, `application/xml`) are served as UTF-8 strings; all others are
returned as `BlobResource`.

```python
from lauren_mcp.server import FileResource, register_file_resource

readme = FileResource("README.md", "file:///project/README.md")
```

---

### HttpResource

```python
# from lauren_mcp.server import HttpResource

class HttpResource:
    def __init__(
        self,
        url: str,
        uri: str,
        *,
        name: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        headers: dict[str, str] | None = None,
        timeout: float = 30.0,
    ) -> None: ...
```

Fetch an HTTP URL and expose the response body as an MCP resource. Requires the
`[http]` extra (`httpx`). MIME type is taken from the response `Content-Type` header
when `mime_type` is omitted. Text-like responses become strings; others become
`BlobResource`.

```python
from lauren_mcp.server import HttpResource

weather = HttpResource(
    "https://api.example.com/weather",
    "weather://current",
    name="current-weather",
    headers={"Authorization": "Bearer token"},
)
```

---

### DirectoryResource

```python
# from lauren_mcp.server import DirectoryResource

class DirectoryResource:
    def __init__(
        self,
        path: str | Path,
        uri: str,
        *,
        name: str | None = None,
        description: str | None = None,
        pattern: str = "*",
        recursive: bool = False,
        include_hidden: bool = False,
    ) -> None: ...
```

List files in a directory and expose the result as a JSON array resource. Returns a
JSON-serialised list of relative file paths matching `pattern`. MIME type is always
`"application/json"`. Hidden files (names starting with `.`) are excluded by default.

```python
from lauren_mcp.server import DirectoryResource

src_files = DirectoryResource(
    "src/",
    "file:///project/src/",
    pattern="**/*.py",
    recursive=True,
)
```

---

### register_file_resource

```python
# from lauren_mcp.server import register_file_resource

def register_file_resource(catalog: McpCatalogManager, resource: FileResource) -> None:
```

Register a `FileResource` with an `McpCatalogManager` instance. Convenience wrapper
around `catalog.register_resource(resource.as_mcp_resource_meta())`.

---

### register_http_resource

```python
# from lauren_mcp.server import register_http_resource

def register_http_resource(catalog: McpCatalogManager, resource: HttpResource) -> None:
```

Register an `HttpResource` with an `McpCatalogManager` instance.

---

### register_directory_resource

```python
# from lauren_mcp.server import register_directory_resource

def register_directory_resource(catalog: McpCatalogManager, resource: DirectoryResource) -> None:
```

Register a `DirectoryResource` with an `McpCatalogManager` instance.

---

### ToolStream

```python
@dataclass
class ToolStream(Generic[T]):
    generator: AsyncGenerator[T, None]
    total: int | None = None
    accumulate: Callable[[list[T]], Any] | None = None
```

Return type for ``@mcp_tool`` methods that produce results incrementally.
Each value yielded by ``generator`` is sent to the client as a
``notifications/progress`` event during the tool call; when the generator is
exhausted the accumulated result becomes the ``tools/call`` response.

**Accumulation:** ``str`` chunks are joined with ``""`` by default; all other
types use the last yielded value (or ``None`` for an empty generator); supply
``accumulate=lambda chunks: ...`` for custom reduction.

**Progress notifications** are only delivered when the client includes a
``progressToken`` in ``_meta`` of the ``tools/call`` request.  Without one the
stream still runs and the accumulated result is returned normally.

```python
from lauren_mcp import ToolStream

@mcp_tool()
async def generate(self, prompt: str) -> ToolStream[str]:
    """Stream completion tokens for *prompt*."""
    async def gen():
        async for token in llm.stream_complete(prompt):
            yield token
    return ToolStream(gen())
```

To produce a numeric total with ``accumulate``:

```python
return ToolStream(
    number_generator(),
    total=expected_count,
    accumulate=lambda chunks: sum(chunks),
)
```

---

### LaurenMcpMemoryServer

Deployable ``@mcp_server`` backed by in-memory storage (subclassable for SQLite).
Exposes tools ``save_conversation``, ``delete_conversation``, ``save_user_fact``,
``get_user_facts``, ``delete_user_fact`` and resource ``memory://conversations/{id}``.
Use with :class:`~lauren_ai.mcp.McpConversationStore` / ``McpUserMemoryStore`` in ``lauren-ai``.

```python
from lauren_mcp import LaurenMcpMemoryServer, McpServerModule
module = McpServerModule.for_root(LaurenMcpMemoryServer)
```

---

### McpExecutionContext

```python
@dataclass(frozen=True)
class McpExecutionContext:
    tool_name: str
    method_name: str
    server_class: type
    headers: Any | None
    execution_context: Any | None
    session_id: str | None
    metadata: dict[str, Any]
    tool_use_id: str | int | None

    def get_metadata(self, key: str, default: Any = None) -> Any: ...
```

Lightweight context passed to per-tool `@use_guards` and `@use_interceptors`.
Built at dispatch time — before the tool method is invoked. Distinct from
`McpToolContext` which is injected into the tool body itself.

`headers` is `None` on stdio and WebSocket (upgrade headers available on WS).
`execution_context` is `None` on WebSocket and stdio; a Lauren
`ExecutionContext` on SSE / Streamable HTTP.

---

### McpForbiddenError

```python
class McpForbiddenError(RuntimeError):
    guard_name: str  # name of the guard that rejected the call
```

Raised inside `make_tools_call_handler` when a per-tool `@use_guards` guard
returns `False`.  The dispatcher catches it and returns `INTERNAL_ERROR`
with `data = {"type": "FORBIDDEN", "guard": guard_name}`.

```python
from lauren import injectable, use_guards
from lauren_mcp import McpForbiddenError, mcp_tool

@injectable()
class AdminGuard:
    async def can_activate(self, ctx: McpExecutionContext) -> bool:
        return ctx.headers.get("x-role") == "admin" if ctx.headers else False

@use_guards(AdminGuard)
@mcp_tool()
async def admin_only(self) -> dict: ...
# Calling this without the x-role header → INTERNAL_ERROR with data.type="FORBIDDEN"
```

---

### McpCallHandler

```python
class McpCallHandler:
    async def handle(self) -> dict[str, Any]: ...
```

Passed to per-tool interceptors as the "next step" in the chain.
Call `await call_handler.handle()` to invoke the next interceptor or ultimately
the tool method itself.  Mirrors `lauren.types.CallHandler` but returns
`dict[str, Any]` (the tools/call result) instead of an HTTP Response.

```python
from lauren import interceptor, use_interceptors
from lauren_mcp import McpCallHandler

@interceptor()
class AuditInterceptor:
    async def intercept(self, ctx: McpExecutionContext, call_handler: McpCallHandler) -> dict:
        result = await call_handler.handle()
        print(f"Tool {ctx.tool_name} completed, isError={result.get('isError')}")
        return result

@use_interceptors(AuditInterceptor)
@mcp_tool()
async def my_tool(self) -> dict: ...
```

---

## Exceptions

These exceptions are not in `__all__` but are commonly caught by callers:

- `McpError` — base class for all lauren-mcp exceptions
- `McpConnectionError` — cannot connect or lost connection
- `McpHandshakeError` — protocol version mismatch during initialize
- `McpToolError` — server returned an error response for a tool call
- `McpToolNotFoundError` — tool name not in server's tool list
- `McpTimeoutError` — operation exceeded the configured timeout
- `JsonRpcParseError` — raw data is not valid JSON-RPC 2.0

---

## Internal modules (not in __all__, for contributors)

- `lauren_mcp._server._dispatcher.McpDispatcher` — body-based JSON-RPC routing
- `lauren_mcp._server._session.SseSessionStore` — HTTP+SSE session lifecycle
- `lauren_mcp._server._ws.WsHandler` — WebSocket connection handler
- `lauren_mcp._server._sse.SseHandler` — HTTP+SSE connection handler
- `lauren_mcp._server._handshake.run_handshake` — MCP initialize sequence
- `lauren_mcp._client._McpBaseRemoteClient` — shared base for all client transports
- `lauren_mcp._client._ws.WsClient` — WebSocket client implementation
- `lauren_mcp._client._http.HttpSseClient` — HTTP+SSE client implementation