fix(examples): parse the canonical EEP gate response in the langgraph agent

examples/langgraph-eep-agent/README.md

@@ -6,8 +6,10 @@
 
 1. **Discovery**: Agent resolves an entity via Layer 1 REST, reads `Link` headers and `/.well-known/eep.json`.
 2. **Subscription**: Agent subscribes to entity events via `POST /eep/subscribe` (Layer 2 webhook delivery).
-3. **Gate handling**: When the agent hits a 402 (payment) or 403 (credential/agreement), it uses `eep-gates-python` to construct and submit gate proofs.
-4. **Event processing**: Incoming webhook events are validated with `eep-validator-python`, signature-checked with `eep-signer-python`, and routed to a LangGraph processing graph.
+3. **Gate handling**: When the agent hits a `402 access_restricted` or `403 access_forbidden` response, it parses the canonical `unmet_requirements[]` (each carries a machine-readable `resolution_hint`) and constructs a matching gate proof **per requirement** — no LLM needed to decide what to do.
+4. **Event processing**: Incoming webhook events are validated against the CloudEvents envelope and signature-checked with a Standard Webhooks HMAC verifier (the algorithm in `eep-signer`), then routed to a LangGraph processing graph.
+
+> This example implements the EEP wire contracts (canonical gate response, Standard Webhooks HMAC) **inline** so it reads as a single self-contained file. In production, import [`eep-gates`](../../packages/eep-gates-python/), [`eep-signer`](../../packages/eep-signer-python/), and [`eep-validator`](../../packages/eep-validator-python/) instead of re-implementing them.
 5. **Claude reasoning**: Each event is summarized and acted on by Claude (via `langchain-anthropic`).
 
 ## Architecture
@@ -61,13 +63,31 @@ python agent.py
 
 ## How gate handling works
 
-When the agent encounters a gated resource:
+Both `402 access_restricted` and `403 access_forbidden` use the same canonical
+body (`gate.402-response.json` / `gate.403-response.json`):
+
+```json
+{
+  "error": "access_restricted",
+  "resource": "content.papers.full_text",
+  "current_tier": "free",
+  "required_tier": "paid",
+  "unmet_requirements": [
+    { "type": "payment", "amount": 0.1, "currency": "USD", "per": "request",
+      "resolution_hint": "Pay $0.10 via the payment_methods URL" }
+  ]
+}
+```
 
-- **402 Payment Required**: The agent reads the `gate_type` and payment requirements from the response body, constructs a payment proof (x402 or on-chain hash), and retries with `gate_proofs` in the request.
-- **403 Forbidden (credential)**: The agent presents a Verifiable Presentation from its credential store.
-- **403 Forbidden (agreement)**: The agent fetches the agreement document, computes its hash, signs it with the agent DID private key, and retries.
+`handle_gate_challenge()` iterates `unmet_requirements` and builds one proof per
+entry, routed on each requirement's `type` (`payment` → `token`, `credential` →
+a VC in the accepted format, `agreement` → a signature over `document_hash`,
+`identity`/`trust`/`connection`, …). Every requirement carries a
+`resolution_hint`, so the agent decides what to satisfy **without** an LLM call.
+Requirement types the demo cannot auto-satisfy (e.g. custom `x-*`) are skipped.
 
-These flows use `eep_gates.proof_validator` and `eep_gates.access_resolver` from the `eep-gates-python` package.
+`test_agent.py` asserts this parsing against the canonical shapes (run
+`python -m pytest test_agent.py`).
 
 ## Related
 

examples/langgraph-eep-agent/agent.py

@@ -74,45 +74,79 @@ def subscribe(subscribe_url: str, entity_did: str, webhook_url: str, api_key: st
     return data
 
 
-def handle_gate_challenge(target: str, entity: str, status: int, body: dict) -> dict | None:
-    """Construct gate proof based on 402/403 response."""
-    if status == 402:
-        gate_type = body.get("gate_type", "payment")
-        print(f"[gate] 402 Payment Required — gate_type={gate_type}", file=sys.stderr)
+def _proof_for_requirement(req: dict) -> dict | None:
+    """Construct a (demo) gate proof matching one unmet requirement.
+
+    Proof shapes mirror @eep-dev/gates `GateProof`. The values here are
+    placeholders — a real agent would pay, present a real Verifiable
+    Credential, or sign the agreement document — but the *structure* and the
+    requirement-type routing are exactly what a publisher's proof verifier
+    expects (see packages/@eep-dev/gates/src/types.ts).
+    """
+    rtype = req.get("type", "")
+    if rtype == "payment":
+        # PaymentRequirement: { amount, currency, per, payment_methods?, x402? }
+        return {"type": "payment", "token": "demo_payment_token", "provider": "demo"}
+    if rtype == "credential":
+        # CredentialRequirement: { credential_type, issuer?, accepted_formats? }
+        fmt = (req.get("accepted_formats") or ["jwt_vc"])[0]
+        return {"type": "credential", "credential": "<demo_jwt_vc>", "format": fmt}
+    if rtype == "agreement":
+        # AgreementRequirement: { document_hash, document_url, signature_algo? }
+        doc_hash = req.get("document_hash", "")
         return {
-            "gate_type": gate_type,
-            "proof": {
-                "type": "payment",
-                "tx_hash": "0x_demo_payment_hash",
-                "amount": body.get("amount", "0.01"),
-                "currency": body.get("currency", "USD"),
-            },
+            "type": "agreement",
+            "document_hash": doc_hash,
+            "signature": f"did:key:agent_demo_sig_{doc_hash[:8]}",
         }
-    elif status == 403:
-        error = body.get("error", "")
-        if error == "agreement_required":
-            print("[gate] 403 Agreement Required — signing agreement hash", file=sys.stderr)
-            doc_hash = body.get("agreement_hash", "")
-            return {
-                "gate_type": "agreement",
-                "proof": {
-                    "type": "agreement",
-                    "document_hash": doc_hash,
-                    "signature": f"did:key:agent_demo_sig_{doc_hash[:8]}",
-                },
-            }
-        elif error == "credential_required":
-            print("[gate] 403 Credential Required — presenting VC", file=sys.stderr)
-            return {
-                "gate_type": "credential",
-                "proof": {
-                    "type": "credential",
-                    "verifiable_presentation": {"holder": "did:key:agent_demo", "proof": "..."},
-                },
-            }
+    if rtype == "identity":
+        return {"type": "identity", "method": req.get("method", "did_verified"), "evidence": "did:key:agent_demo"}
+    if rtype == "trust":
+        return {"type": "trust", "self_attested": True}
+    if rtype == "connection":
+        return {"type": "connection", "subscriber_did": "did:key:agent_demo", "relation": req.get("relation", "follower")}
+    # Unknown / custom x-* requirement: nothing the demo can auto-satisfy.
     return None
 
 
+def handle_gate_challenge(status: int, body: dict) -> dict | None:
+    """Parse a canonical EEP 402/403 gate response and build matching proofs.
+
+    The canonical body (`gate.402-response.json` / `gate.403-response.json`) is:
+
+        { "error": "access_restricted" | "access_forbidden",
+          "resource": "...", "current_tier": "...", "required_tier": "...",
+          "unmet_requirements": [ { "type": ..., "resolution_hint": ..., ... } ],
+          "available_tiers"?: {...} }
+
+    It is NOT a flat ``{gate_type, amount, currency}`` object. *Every* gate type
+    (payment, credential, agreement, identity, ...) arrives as an entry in
+    ``unmet_requirements``, each carrying a machine-readable ``resolution_hint``
+    so the agent needs no LLM to decide what to do. We build one proof per
+    requirement we can satisfy and return them together for the retry.
+    """
+    if body.get("error") not in ("access_restricted", "access_forbidden"):
+        return None
+
+    proofs: list[dict] = []
+    for req in body.get("unmet_requirements", []) or []:
+        hint = req.get("resolution_hint", "")
+        print(f"[gate] HTTP {status} unmet '{req.get('type')}': {hint}", file=sys.stderr)
+        proof = _proof_for_requirement(req)
+        if proof:
+            proofs.append(proof)
+
+    if not proofs:
+        print("[gate] no auto-satisfiable requirements in challenge", file=sys.stderr)
+        return None
+
+    return {
+        "resource": body.get("resource"),
+        "required_tier": body.get("required_tier"),
+        "proofs": proofs,
+    }
+
+
 # ─── Webhook Signature Verification ──────────────────────────────────────────
 
 
@@ -338,9 +372,9 @@ def main():
         except Exception:
             pass
 
-        proof = handle_gate_challenge(EEP_TARGET, "u/acme-corp", e.response.status_code, body)
+        proof = handle_gate_challenge(e.response.status_code, body)
         if proof:
-            print(f"[gate] Would retry with proof: {json.dumps(proof, indent=2)}", file=sys.stderr)
+            print(f"[gate] Would retry with proofs: {json.dumps(proof, indent=2)}", file=sys.stderr)
         else:
             print(f"[error] HTTP {e.response.status_code}: {e}", file=sys.stderr)
         sys.exit(1)

examples/langgraph-eep-agent/requirements.txt

@@ -2,6 +2,9 @@ langgraph>=0.2
 langchain>=0.3
 langchain-anthropic>=0.3
 httpx>=0.27
-eep-gates
-eep-signer
-eep-validator
+# The example inlines the EEP wire contracts (canonical gate response,
+# Standard Webhooks HMAC) so it reads as one self-contained file. For
+# production, prefer these packages over re-implementing them:
+#   eep-gates       # gate config / access resolution / 402 builder
+#   eep-signer      # Standard Webhooks HMAC sign + verify
+#   eep-validator   # SSRF + event-type pattern checks

examples/langgraph-eep-agent/test_agent.py

@@ -0,0 +1,84 @@
+# Copyright 2026 EEP Contributors — Apache-2.0
+"""Unit tests for the LangGraph EEP example's gate-challenge parser.
+
+These assert the example parses the *canonical* EEP gate response
+(`schemas/v0.1/gate.402-response.json` / `gate.403-response.json`) — an
+`unmet_requirements[]` array — rather than a flat `{gate_type, amount,
+currency}` object, which earlier revisions wrongly assumed.
+
+Run from this directory: ``python -m pytest test_agent.py``
+"""
+
+import agent
+
+
+def test_parses_canonical_402_payment():
+    body = {
+        "error": "access_restricted",
+        "resource": "content.papers.full_text",
+        "current_tier": "free",
+        "required_tier": "paid",
+        "unmet_requirements": [
+            {
+                "type": "payment",
+                "amount": 0.1,
+                "currency": "USD",
+                "per": "request",
+                "resolution_hint": "Pay $0.10 via the payment_methods URL",
+            }
+        ],
+    }
+    result = agent.handle_gate_challenge(402, body)
+    assert result is not None
+    assert result["resource"] == "content.papers.full_text"
+    assert result["required_tier"] == "paid"
+    assert len(result["proofs"]) == 1
+    assert result["proofs"][0]["type"] == "payment"
+    # PaymentProof requires a `token` field.
+    assert "token" in result["proofs"][0]
+
+
+def test_parses_multiple_unmet_requirements():
+    body = {
+        "error": "access_restricted",
+        "resource": "x",
+        "current_tier": "public",
+        "required_tier": "verified",
+        "unmet_requirements": [
+            {"type": "agreement", "document_hash": "sha256:deadbeefcafe", "document_url": "https://x/doc"},
+            {"type": "credential", "credential_type": "AcademicAffiliation", "accepted_formats": ["ldp_vc"]},
+        ],
+    }
+    result = agent.handle_gate_challenge(402, body)
+    by_type = {p["type"]: p for p in result["proofs"]}
+    assert set(by_type) == {"agreement", "credential"}
+    assert by_type["credential"]["format"] == "ldp_vc"
+    assert by_type["agreement"]["document_hash"] == "sha256:deadbeefcafe"
+
+
+def test_handles_403_access_forbidden():
+    body = {
+        "error": "access_forbidden",
+        "resource": "x",
+        "current_tier": "public",
+        "required_tier": "member",
+        "unmet_requirements": [{"type": "identity", "method": "kyc"}],
+    }
+    result = agent.handle_gate_challenge(403, body)
+    assert result["proofs"][0]["type"] == "identity"
+    assert result["proofs"][0]["method"] == "kyc"
+
+
+def test_ignores_non_gate_error():
+    assert agent.handle_gate_challenge(500, {"error": "internal_error"}) is None
+
+
+def test_no_autosatisfiable_requirements_returns_none():
+    body = {
+        "error": "access_restricted",
+        "resource": "x",
+        "current_tier": "a",
+        "required_tier": "b",
+        "unmet_requirements": [{"type": "x-custom-thing"}],
+    }
+    assert agent.handle_gate_challenge(402, body) is None