Langchain MCP Secure

langchain mcps

](https://pypi.org/project/langchain-mcpsecure/) 

LangChain的MCPS（MCP安全）集成 --AI代理的加密身份和信任验证。

使用一行代码为任何LangChain代理或链添加零信任身份验证。

安装

pip install langchain-mcpsecure

从源代码构建

要求：Python 3.9+， git

# Clone the repository
git clone https://github.com/Anakintano/langchain-mcp-secure.git
cd langchain-mcp-secure

# Install in editable mode with all dependencies
pip install -e .

# Install development tools (testing, linting)
pip install pytest pytest-cov pylint mypy bandit

# Run tests to verify the build
pytest tests/ -v --cov=langchain_mcps

所需库：

• mcp-secure>=1.0.0 --ECDSA P-256护照签名（自动安装）
• langchain-core>=0.2.0 --LangChain回调挂钩（自动安装）

看 贡献.md 用于依赖选择策略和贡献者指南。

______________________________________________________________________

快速开始

回调处理程序（推荐）

通过回调连接到任何LangChain代理或链：

from mcp_secure import generate_key_pair, create_passport, sign_passport
from langchain_mcps import MCPSCallbackHandler

# Generate keys and create a signed passport
authority_keys = generate_key_pair()
agent_keys = generate_key_pair()
passport = create_passport(
    name="my-agent",
    version="1.0.0",
    public_key=agent_keys["public_key"],
)
signed_passport = sign_passport(passport, authority_keys["private_key"])

# Create the handler
handler = MCPSCallbackHandler(
    passport=signed_passport,
    authority_public_key=authority_keys["public_key"],
    private_key=agent_keys["private_key"],  # optional: signs actions
)

# Use with any LangChain chain or agent
result = my_chain.invoke(
    {"question": "What is MCPS?"},
    config={"callbacks": [handler]},
)

# Check verification status and audit log
print(handler.is_verified)  # True
print(handler.audit_log)    # [{timestamp, event, action, ...}, ...]

中间件包装器

用验证门包裹任何LangChain Runnable：

from langchain_mcps import with_mcps

secure_chain = with_mcps(my_chain, signed_passport, authority_keys["public_key"])
result = secure_chain.invoke({"question": "hello"})
# Raises PermissionError if passport is invalid, expired, or revoked

特性

• 身份验证 --在采取任何代理行动之前，ECDSA P-256护照验证
• 行动签名 --对每个链/工具调用进行加密签名
• 信任级别 --强制执行最低信任（L0未签名到L4已审核）
• 撤销检查 --通过AgentSign信任机构进行可选的实时撤销
• Merkle链审计日志 （v2.1）——使用SHA256链哈希进行加密防篡改审计跟踪
• 有时限的临时权限 （v2.2）--通过时间窗口和事件门授予临时工具访问权限
• 代表团链 （v2.3）——具有权限升级防止的代理间授权
• 审计跟踪 --验证/拒绝事件的完整日志，包括委派事件
• 重放保护 --基于随机数的重放攻击防御（护照+委托令牌）
• 零配置 --适用于任何LangChain Runnable（链、代理、工具）

安全架构

langchain mcps提供了5个渐进式安全层，每个层都建立在前一层之上：

┌─────────────────────────────────────────────────────┐
│  v2.3: Delegation Chains                            │
│  ↑ Transitive trust (Agent A → Agent B → Tool)      │
├─────────────────────────────────────────────────────┤
│  v2.2: Time-Bound Ephemeral Permissions             │
│  ↑ When can you act? (time windows + approval gates)│
├─────────────────────────────────────────────────────┤
│  v2.1: Merkle-Chain Audit Log                       │
│  ↑ What did you do? (tamper-proof, non-repudiation) │
├─────────────────────────────────────────────────────┤
│  v2.0: Capability-Scoped Passports                  │
│  ↑ What can you do? (least-privilege constraints)   │
├─────────────────────────────────────────────────────┤
│  v1.0: Passport Identity & Verification             │
│  ↑ Who are you? (ECDSA P-256 signature)             │
└─────────────────────────────────────────────────────┘

每一层都添加了安全保证：

• v1.0： 身份验证——你是谁？
• v2.0: 授权——你能做什么？
• v2.1: 审计——你做了什么？
• v2.2: 有时间限制——你什么时候能做？
• v2.3: 代理人——你能授权其他人吗？

信任级别

API

MCPSCallbackHandler

属性和方法（v2.1）

Merkle链审计追踪：

# Get the merkle root (single tamper-evident hash of entire audit trail)
root = handler.merkle_root  # str or None

# Verify the entire audit chain is intact
is_valid = handler.verify_audit_chain()  # bool

# Sign the merkle root for external publication
signed_root = handler.sign_merkle_root()  # dict with signature

# Access full audit log with cryptographic hashes
log = handler.audit_log  # [{timestamp, event, passport_id, action, entry_hash, previous_entry_hash, ...}, ...]

示例：导出篡改检测服务的默克尔根

handler = MCPSCallbackHandler(
    passport=signed_passport,
    authority_public_key=authority_keys["public_key"],
    private_key=agent_keys["private_key"],
    on_merkle_root_finalized=lambda root, signature: print(f"Root: {root}\nSig: {signature}"),
)

# ... run agent/chain ...

# Sign and export merkle root
signed_root = handler.sign_merkle_root()
# Send signed_root to IPFS, blockchain, or audit service for tamper-evident proof

with_mcps(chain, passport, authority_public_key, **kwargs)

便利包装。返回一个 MCPSChainWrapper 和 .invoke(), .ainvoke(), .stream(), .batch().

______________________________________________________________________

DelegationToken （v2.3）

RFC 8693 JWT携带代理到代理的委托授权。

令牌字段： iss, sub, aud, iat, exp, jti, act, capabilities, parent_passport_id, delegation_depth, max_delegation_depth

______________________________________________________________________

DelegationTokenValidator （v2.3）

具有防重放功能的6步验证门。

DelegationVerificationResult 领域： valid: bool, reason: str, token: DelegationToken | None

______________________________________________________________________

QuotaPool （v2.3）

共享滑动窗口速率限制器。父代理的所有代表共享每个工具的一个预算。

______________________________________________________________________

intersect_capabilities(delegator_caps, requested_caps) → dict （v2.3）

计算能力交集。结果总是 delegator_caps --防止升级。

• allowed_tables：设置交点
• rate_limit: min(delegator, requested) 每个字段
• 其他约束：委托人的价值获胜

______________________________________________________________________

v2.2:有时限的短暂权限

授予基于时间窗口的临时工具访问权限。工具可以限制在特定的时间间隔内（例如，维护窗口、承包商访问、紧急升级）。

Passport功能架构（v2.2）

signed_passport["capabilities"] = {
    "database_write": {
        "allowed": True,
        "constraints": {},
        "permission_windows": [
            # Only allowed Saturday 2am-4am UTC
            {"start_time": 1711324800.0, "end_time": 1711332000.0},
        ],
    },
    "send_email": {
        "allowed": True,
        "constraints": {"recipient_domains": ["example.com"]},
        # No permission_windows = always allowed during valid passport
    },
    "escalated_delete": {
        "allowed": True,
        "constraints": {},
        "permission_gates": [
            {"gate_type": "manual_approval", "config": {"approval_required": True}}
        ],
    },
    "file_delete": {"allowed": False},  # Explicitly forbidden
}

时间窗口

• 时间间隔： [start_time, end_time) --包容性开始，排他性结束
• 逻辑： OR——如果当前时间落在任何窗口中，则允许代理
• 无窗口字段： 始终允许使用工具（向后兼容）
• 空窗口列表 []: 绝不允许使用工具

handler = MCPSCallbackHandler(
    passport=signed_passport,
    authority_public_key=authority_keys["public_key"],
    current_time_provider=lambda: time.time(),  # injectable for testing
)

权限门

盖茨需要外部决定才能调用工具。这 on_permission_gate_triggered 回调必须返回 (is_allowed: bool, reason: str).

def my_approval_gate(tool_name, gate_config):
    approved = approval_service.check(tool_name)
    return approved, "approved" if approved else "awaiting_approval"

handler = MCPSCallbackHandler(
    passport=signed_passport,
    authority_public_key=authority_keys["public_key"],
    on_permission_gate_triggered=my_approval_gate,
)

隐式拒绝（v2.0+）

任何工具 未列出 在v2.0+版本的护照功能中，dict会被自动拒绝。

向后兼容

• v1.0护照（无功能）允许随时使用所有工具
• v2.0/v2.1护照无 permission_windows 随时允许使用工具
• 只有护照 permission_windows 执行时间限制

______________________________________________________________________

v2.1:Merkle链审计日志

每个审计条目都通过SHA256哈希与前一个条目加密链接，形成一个防篡改链。这 默克尔根 （最后一个条目的哈希值）是一个可以签名和发布的单个值，以证明审计跟踪没有被修改。

运作原理

Entry 1: {timestamp, event, ...} → hash(entry_1) = hash_1
Entry 2: {timestamp, event, previous_hash: hash_1, ...} → hash(entry_2 + hash_1) = hash_2
Entry 3: {timestamp, event, previous_hash: hash_2, ...} → hash(entry_3 + hash_2) = hash_3 ← merkle_root

如果任何条目被篡改，链条就会断裂 verify_audit_chain() 返回False。

用例

1. 合规性审计 --导出merkle根以证明审计跟踪的完整性
2. 篡改检测 --呼叫 verify_audit_chain() 检测修改
3. 第三方验证 --签署merkle根并发布（IPFS、区块链等）
4. 法医证据 --日志未被更改的加密证明

向后兼容

v2.1与v1.0完全向后兼容：

• v1.0护照（无功能）仍然有效
• v1.0审核日志仍可通过以下方式访问 handler.audit_log
• v2.1只是在每个条目中添加哈希字段

______________________________________________________________________

v2.3委托链（代理到代理授权）

代理A可以使用有时间限制的、有范围的委托令牌安全地将有限的工作委托给代理B。B不能升级到A的权限之外——这是由令牌结构本身在协议级别强制执行的。

授权如何运作

代币创建、展示、验证和执行分为4个阶段：

PHASE 1: Creation          PHASE 2: Present           PHASE 3: Verify            PHASE 4: Execute & Audit
──────────────────         ─────────────────          ────────────────           ─────────────────────────
Agent A                    Agent B                    MCPSCallbackHandler        System
  │                          │                           │                           │
  ├─ Create JWT token        │                           │                           │
  │  ├─ iss: agent-a         │                           │                           │
  │  ├─ sub: agent-b         │                           │                           │
  │  ├─ exp: +30 min         │                           │                           │
  │  ├─ jti: abc123          │                           │                           │
  │  └─ capabilities:        │                           │                           │
  │     {database_read:      │                           │                           │
  │      [customers]}        │                           │                           │
  │                          │                           │                           │
  ├─ Sign (ECDSA P-256)      │                           │                           │
  │                          │                           │                           │
  └─────── token_jwt ───────→├─ Invoke tool with:       │                           │
                             │  ├─ B's passport         │                           │
                             │  ├─ delegation_token     │                           │
                             │  └─ tool_call            │                           │
                             │                          │                           │
                             └──────────────────────────┤                           │
                                                        ├─ STEP 1: B's passport ✓  │
                                                        ├─ STEP 2: JWT structure ✓  │
                                                        ├─ STEP 3: A's sig ✓        │
                                                        ├─ STEP 4: TTL/nonce ✓      │
                                                        ├─ STEP 5: caps ⊆ A's ✓     │
                                                        ├─ STEP 6: chain depth ✓    │
                                                        │  → ALLOW                  │
                                                        │                           ├─ Execute tool
                                                        │                           │  (database_read,
                                                        │                           │   customers)
                                                        │                           │
                                                        │                           ├─ Append to audit chain:
                                                        │                           │  {action: delegation_used,
                                                        │                           │   agent_b, delegated_from: a,
                                                        │                           │   jti: abc123,
                                                        │                           │   entry_hash: sha256(...)}
                                                        │                           │
                                                        │                           └─ Merkle linked to
                                                        │                              previous_hash

验证门（第3阶段）是安全关键路径：6次连续检查，任何故障都会引发 PermissionError 并在调用该工具之前将拒绝记录到审计链中。

快速示例

from mcp_secure import generate_key_pair, create_passport, sign_passport
from langchain_mcps import MCPSCallbackHandler
from langchain_mcps.delegation import DelegationToken

# Agent A creates a delegation token for Agent B
delegator_caps = {
    "database_read": {
        "allowed": True,
        "constraints": {"allowed_tables": ["customers", "orders", "payments"]},
    }
}

# B gets access only to "customers" — subset of A's tables
token = DelegationToken.create(
    delegator_agent_id=agent_a_passport["passport_id"],
    delegatee_agent_id=agent_b_passport["passport_id"],
    delegator_capabilities=delegator_caps,
    requested_capabilities={
        "database_read": {
            "allowed": True,
            "constraints": {"allowed_tables": ["customers"]},
        }
    },
    ttl_seconds=1800,   # 30 min
)
token_jwt = token.to_jwt(agent_a_private_key)

# Agent B uses the delegation token
handler = MCPSCallbackHandler(
    passport=agent_b_passport,
    authority_public_key=authority_public_key,
    delegation_token_jwt=token_jwt,
    delegator_passport=agent_a_passport,   # provides A's public key
)
handler.on_tool_start({"name": "database_read"}, "SELECT * FROM customers")
# Tool executes — delegation verified, action audited

6步验证门

每次工具调用都要经过六次安全检查：

步骤5是关键的安全保证：令牌本身携带 intersection(A's caps, requested scope)如果B请求表a没有，则交叉点为空，请求被拒绝——无论B自己的护照上写着什么。

共享配额池

A的所有代表共享一个费率限制池，防止任何一位代表耗尽家长的预算：

# A has rate_limit=100/hour for database_read
# A delegates to B and C (both use parent pool key)

# B uses 60 calls  → 40 remaining in pool
# C uses 40 calls  → 0 remaining
# Any further calls by B or C are rejected

审计跟踪

所有委托事件都在现有的AuditChain中被merkle链接：

delegation_verified  →  tool_start  →  chain_end
       │                    │               │
    entry_hash ──────── prev_hash    prev_hash

呼叫 handler.verify_audit_chain() 证明完整的委托线索是显而易见的。

安全特性

设计参考

基于经过验证的授权模式构建：

• AWS IAM策略交集 --特权升级预防
• RFC 8693令牌交换 --标准化的JWT委托格式（由Azure AD、ZITADEL使用）
• SAML断言ID随机数缓存 --防止重放

______________________________________________________________________

安全特性

每一层都可以防止不同类别的威胁：

Layer              Threat Prevented         Mechanism                          Status
────────────────────────────────────────────────────────────────────────────────────
v1.0  Identity     Impersonation            ECDSA P-256 signature              ✅ Prevents
                                            Only the passport holder can
                                            produce a valid signature

v2.0  Capability   Privilege escalation     Implicit deny + constraints        ✅ Prevents
                                            Tool not listed → rejected
                                            allowed_tables enforced per call

v2.1  Audit        Tampering / denial       SHA256 merkle-chain linking        ✅ Prevents
                                            Modify any entry → chain breaks
                                            verify_chain() returns False

v2.2  Time-bound   Unauthorized timing      Time window + gate enforcement     ✅ Prevents
                                            Outside window → rejected
                                            Gate callback must approve

v2.3  Delegation   Over-delegation          Capability intersection            ✅ Prevents
                                            B's perms ⊆ A's perms (protocol)
                                            B cannot escalate beyond A

威胁模型覆盖范围

• ✅ 冒充 --代理不能伪造身份（ECDSA签名）
• ✅ 权限提升 --代理不能超过授予的权限（交叉）
• ✅ 篡改 --审计跟踪不能修改（merkle链）
• ✅ 未经授权的计时 --代理无法在允许的窗口外执行操作（时间检查）
• ✅ 过度授权 --委托人授予的权限不能超过委托人拥有的权限（约束交集）
• ✅ 代币伪造 --委托令牌需要委托人的私钥（ECDSA P-256）
• ✅ 重放攻击 --JTI随机缓存+TTL（护照+委托令牌）

______________________________________________________________________

路线图

v2.4--企业规模委托

多进程部署和代理编排的延迟功能。

\[v2.4.1\]分布式配额和非存储（Redis后端）· #9

状态： 计划中· 努力 1-2周

当前 QuotaPool 和 DelegationTokenValidator._nonce_cache 它们位于内存中，并且是本地进程。 在多副本部署中，不同的进程可以接受重放的令牌。

• Redis支持 QuotaPool （原子 ZADD + ZCOUNT，在重新启动后仍然有效）
• Redis支持的nonce缓存（原子 SET NX EX，跨流程共享）
• 如果Redis不可用，则自动回退到内存中

\[v2.4.2\]多跳委派（A→B→C链）· #10

状态： 计划中· 努力 2-3周· 被阻止： #9

启用代理编排，代理B可以安全地将其工作的一个子集委托给代理C。

• 循环检测——防止A→B→A循环（DFS结束 delegation_chain 现场）
• 防止升级——B委托给C时的上限必须是⊆A最初授予B的上限
• 可配置的最大深度（MVP默认为1，计划最多3跳）

v2.5--令牌本机权限和异步

细粒度的每个令牌控制和非阻塞委托验证。

\[v2.5.1\]令牌中的时间窗口和权限门· #11

状态： 计划中· 努力 1-2周

允许A嵌入显式 permission_windows 和 permission_gates 在令牌本身内部，独立于A的护照。示例：“B可以使用 database_read 仅在今晚02:00至04:00之间，并且只有在审批者webhook同意的情况下。"

• 代币创建时的交叉点（B不能获得比a护照允许的更宽的窗口）
• 验证门步骤5中添加了窗口+门检查

\[v2.5.2\]异步委托门（ainvoke支持）· #12

状态： 计划中· 努力 3-4天

MCPSChainWrapper.ainvoke() 当前调用同步 _gate()，阻塞事件循环。

• DelegationTokenValidator.verify_async() --非阻塞6步验证
• QuotaPool.check_and_decrement_async() --异步配额会计（受益于v2.4.1 Redis）
• MCPSChainWrapper._gate_async() 连接到 ainvoke(), astream(), abatch()

贡献

所有路线图功能都对社区贡献开放。

1. 对相关问题发表评论以提出索赔
2. 尽早打开PR草案，以便我们能够就方法达成一致
3. 在公关描述中参考问题和路线图

当前测试基线： 137次测试·90%覆盖率 --新功能应该保持或改进这一点。

______________________________________________________________________

项目治理