Metadata-Version: 2.4
Name: mcp-agentlock
Version: 0.2.1
Summary: AgentLock authorization middleware for MCP (Model Context Protocol) servers.
Project-URL: Homepage, https://agentlock.dev
Project-URL: Source, https://github.com/webpro255/mcp-agentlock
Author: AgentLock contributors
License-Expression: AGPL-3.0-or-later
License-File: LICENSE
Keywords: agent,agentlock,ai,authorization,mcp,model-context-protocol
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: agentlock>=1.7
Requires-Dist: mcp<2,>=1.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Description-Content-Type: text/markdown

# mcp-agentlock

[![License: AGPL v3](https://img.shields.io/badge/License-AGPL_v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
[![Tests](https://img.shields.io/badge/tests-66%20passing-brightgreen)](#proof)
[![agentlock.dev](https://img.shields.io/badge/site-agentlock.dev-0a0a0a)](https://agentlock.dev)

Per-tool authorization for MCP servers. Every tool call gated, logged, and bound to a single-use token.

## The Problem

MCP standardises how a server advertises tools and how a client invokes them. It does not standardise who may invoke what. A server that lists a tool has offered it to whatever client holds the connection, on whatever arguments the model produces. `mcp-agentlock` inserts an authorization gate between the dispatch and the tool body, so that a call is evaluated on identity, role, scope, rate, and the provenance of its own parameters before it runs.

## Install

```bash
pip install mcp-agentlock
```

## Quick Start

```python
from typing import Any

import mcp.types as types
from agentlock import AgentLockPermissions, AuthorizationGate, ContextSource
from agentlock.schema import LineagePolicyConfig
from mcp.server.lowlevel import Server

from mcp_agentlock import ToolGuard, lock_call_tool

gate = AuthorizationGate()
server: Server[Any] = Server("my-server")

@lock_call_tool(
    server,
    gate,
    tools={
        "web_fetch": ToolGuard(
            AgentLockPermissions(risk_level="low", allowed_roles=["analyst"]),
            context_source=ContextSource.WEB_CONTENT,   # the lever
        ),
        "send_email": ToolGuard(
            AgentLockPermissions(
                risk_level="high",
                allowed_roles=["analyst"],
                rate_limit={"max_calls": 5, "window_seconds": 60},
                lineage_policy=LineagePolicyConfig(
                    enabled=True,
                    param_lineage_enabled=True,
                    param_lineage_action="deny",
                ),
            )
        ),
    },
    identity_resolver=resolve_identity,   # see examples/basic_server.py
)
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
    ...
```

`lock_call_tool` is a drop-in for `@server.call_tool()`. The guard sits inside it, so the SDK's input validation and result normalisation still run on the outside and a denial comes back as an ordinary `isError` result rather than a dropped connection.

A tool with no entry in `tools` and no `"*"` fallback is **denied**, not dispatched. Deny by default applies to the dispatch table itself: a tool added to the server but forgotten here is unreachable, never unprotected.

## The Source Lever

The one field that turns provenance tracking into enforcement:

```python
"web_fetch": ToolGuard(perms, context_source=ContextSource.WEB_CONTENT)
```

`context_source` declares what kind of content a tool returns. It defaults to `TOOL_OUTPUT`, which the gate resolves to `DERIVED` authority and which changes nothing. Pass `WEB_CONTENT`, `RETRIEVED_DOCUMENT`, or `PEER_AGENT` for a tool whose output an attacker can reach, and the gate resolves those to `UNTRUSTED`, which is what arms the lineage checks against that tool's output.

Trusted by default, untrusted opt-in. Adding the guard to an existing server changes no behaviour until you pull this lever.

## Proof

The end-to-end tests drive a real `mcp.server.lowlevel.Server` over the SDK's in-memory transport: client `call_tool` request, SDK input validation, guarded handler, tool body, ingestion write, SDK result normalisation, client response. No network, no mocks of the SDK.

`test_source_lever_flips_allow_to_deny` runs that path twice. Identical servers, identical calls, identical parameters. The only difference is the `context_source` declared on the fetch tool, and the test asserts the two outcomes **differ** rather than asserting each one separately. Rendered from the same test helpers:

```
--- fetch context_source = TOOL_OUTPUT (default, trusted)
    provenance authority : derived
    sink send_email(to=PAYLOAD) isError = False
    sink output: sent to attacker-drop-box@evil-example.test

--- fetch context_source = WEB_CONTENT (lever pulled)
    provenance authority : untrusted
    sink send_email(to=PAYLOAD) isError = True
    sink output: Tool 'send_email' denied (decision=deny, reason=param_lineage).
                 Parameter 'to' carries a value that originated in untrusted
                 context (web_fetch:cprov_eeaacc5dcd1...)
```

The denial cites the provenance id of the entry that caused it, so the origin and the consequence join in the audit log.

```bash
pytest        # 72 passed, 2 skipped
```

The two skips are the pre-threading rows in `tests/test_carriage.py`, which go dormant once the adapter threads a call's arguments to the ingestion write. They are mirrored against the live rows rather than deleted, so the file measures the change from both sides.

## Boundaries and Security Notes

Three things to know before deploying this.

### 1. Enforcement is cross-hop, bounded by carriage

Lineage catches a value that flows from an untrusted tool's output directly into a later tool call, and it also catches one laundered through an intermediate hop. The guard passes each call's `arguments` to the ingestion write, so the engine's containment linker records a **cross-hop parent link** whenever a call's parameters carry a prior entry's content. Decision-time checks walk those recorded links. Tool A returns untrusted text, tool B is called with that text and emits a rewritten form, tool C is called with the rewritten value: C is denied with reason `param_lineage`, and the denial cites **B's** provenance entry, the relay the value came through, rather than A's.

The boundary is carriage, not hop count. Three things still bound it:

- **An uncarried value produces no link.** If B is invoked without A's content, nothing was carried, no parent is recorded, and the rewritten value reaches C. This is what `test_two_hop_laundering_slips` pins, and it is why that test still asserts the laundered value is delivered.
- **Linking requires whole-content carriage.** A prior entry's entire recorded content has to appear inside one of the call's argument values and clear the engine's containment floor. A partial quotation, or a summary the model composed rather than passed through, does not link.
- **Encoded forms are measured at the engine, not here.** No encoded corpus runs through this adapter, so no adapter-level claim about encoded values is made. See the engine's own limitations record.

Both directions are pinned by `tests/test_carriage.py`, which measures the carried session against the uncarried one rather than asserting either alone. Same behavior as `crewai-agentlock`.

### 2. Lineage is scoped per user_id, not per connection

The gate resolves the session it evaluates with `get_by_user(user_id)`, and holds one active session per user. Two MCP connections that authenticate as the same `user_id` therefore share one provenance context: content that enters context on one connection can deny a call on the other.

For single-user servers and for deployments where distinct callers have distinct `user_id`s, this is what you want, and taint follows the user across reconnects. It is a consideration when one user runs several independent agent sessions concurrently, since those sessions are not isolated from each other. Deployments that need per-session isolation should map each session to its own `user_id` rather than relying on the MCP connection id, which the gate does not consult.

### 3. bind_session is a trust boundary

**This is the single most important operational rule in the package.** Call `bind_session` only from the layer that authenticates the connection: a server lifespan, ASGI middleware that has already validated a bearer token, or a controlled test harness. Never from a tool body, and never from anything the model can reach.

The bound identity is the one input the agent must not be able to choose. An agent that can call `bind_session` can rebind itself to another `user_id` and step outside the provenance context that its own earlier calls contaminated, which discards the lineage enforcement described above. The same applies to any `identity_resolver` that derives identity from tool `arguments`: those are model-controlled, which is why no built-in resolution step reads them.

## Sessions

MCP does not expose a session id to a tool call. `ServerSession` carries no id attribute; the `mcp-session-id` that streamable HTTP mints lives on `StreamableHTTPServerTransport`, which is not reachable from the request context; stateless HTTP passes `None`; stdio has no session concept. So the connection id is derived here, from the identity of the `ServerSession` object the SDK creates once per connection, held in a `WeakKeyDictionary` that drops it when the connection closes.

That id is then **mapped** to an AgentLock session, which is the load-bearing part. `AuthorizationGate.authorize` resolves the session it evaluates with `get_by_user(user_id)` and reads *that* session's provenance log. An ingestion write addressed to the MCP connection id instead would file the evidence where no lineage check ever looks, and every enforcement test would pass vacuously against an empty log. `bind_session` creates the mapping; the guard resolves the write target the same way the gate does, so the two can never disagree.

```python
from mcp_agentlock import bind_session

# From the layer that authenticates the connection. Never from a tool body.
bind_session(gate, user_id="analyst-1", role="analyst")
```

Identity resolution precedence, highest first:

1. `identity_resolver(tool_name, arguments, request_context)`
2. the `SessionBinding` attached to this MCP connection
3. the `AuthContext` set by an enclosing `agentlock_session()`
4. the guard's `default_*` arguments
5. empty, which denies anything requiring auth

`arguments` is passed to the resolver but never read for identity by any built-in step: it is the one input the model controls.

## Execution Reporting

`AuthorizationGate.execute` is synchronous and MCP handlers are coroutines, so this adapter does not hand the gate its execution. It consumes the single-use token itself and reports the attempt and the outcome through `begin_execution` / `confirm_execution`, the engine's own path for callers that own their executor. The audit trail is equivalent; records carry `reported_by="caller"` rather than `"gate"`. Failures are recorded with `status="failed"` and the exception type, then re-raised unchanged.

## License

This package is AGPL-3.0-or-later. See [LICENSE](LICENSE) for the full text.

It depends on [AgentLock](https://github.com/webpro255/agentlock), which is
licensed AGPL-3.0-or-later from v1.3.0 onward (verified at tags v1.2.1, the
last Apache-2.0 line, and v1.3.0 and v1.5.0, both AGPL; the 1.7.0 floor below
verified as AGPL-3.0-or-later from the published PyPI wheel's
License-Expression metadata), and this package requires agentlock>=1.7. Both
sides of the dependency are therefore AGPL: if you run
this in a network service or distribute software built on it, the AGPL's terms
apply to the combined work, including its source-availability requirement for
users who interact with it over a network. Commercial licenses that remove the
AGPL obligations are available at licensing@agentlock.dev.

Read the AGPL and take your own advice on what it requires of you. This note is
a pointer, not legal advice.
