feat(examples): SEC Fillings mcp server is added based on EDGAR Data …

examples/edgar-eep-mcp-python/README.md

@@ -0,0 +1,23 @@
+# SEC EDGAR EEP-MCP Python Example
+
+This example demonstrates gating company facts retrieval from the **SEC EDGAR API** using the **Entity Engagement Protocol (EEP)**.
+
+**EDGAR** should be utilized with your identity and contact information in [Client Demo](./client_demo.py) the `USER_AGENT` should be updated for using this.
+
+## Quick Start
+
+1. Install dependencies and EEP Python bridge:
+   ```bash
+   pip install -r requirements.txt
+   pip install -e ../../packages/eep-mcp-bridge-python
+   ```
+
+2. Run the provider server:
+   ```bash
+   python eep_provider.py
+   ```
+
+3. Call the client demo in another terminal:
+   ```bash
+   python client_demo.py
+   ```

examples/edgar-eep-mcp-python/client_demo.py

@@ -0,0 +1,75 @@
+"""
+EEP Client calling gated SEC Edgar tools.
+Handles 402 Payment Required challenge with mock payment.
+"""
+
+from __future__ import annotations
+
+import json
+import httpx
+
+BRIDGE_URL = "http://localhost:3005"
+USER_AGENT = "Test-Project info@test.com"
+
+def main() -> None:
+    client = httpx.Client(base_url=BRIDGE_URL, timeout=60.0)
+    print("Connecting to SEC Edgar EEP-MCP Bridge...")
+
+    # 1. Manifest Discovery
+    r = client.get("/.well-known/eep.json")
+    print(f"Discovered Manifest! Publisher DID: {r.json()['did']}")
+
+    # 2. Service Catalog Introspection
+    r = client.get("/eep/services")
+    tools = [s["id"] for s in r.json()["services"]]
+    print(f"Introspected {len(tools)} tools: {', '.join(tools)}\n")
+
+    # 3. Call the gated facts tool
+    payload = {
+        "name": "get_company_facts_tool",
+        "arguments": {
+            "ticker": "AAPL",
+            "user_agent":USER_AGENT
+        }
+    }
+
+    print("=== Step 1: Attempt to Call SEC Tool ===")
+    r = client.post("/mcp/tools/call", json=payload)
+    print(f"Status: {r.status_code}")
+
+    if r.status_code == 402:
+        error_body = r.json()
+        print(f"Access Restricted! Required Tier: {error_body['required_tier']}")
+        unmet = error_body["unmet_requirements"]
+        print(f"   Unmet requirements: {[u['type'] for u in unmet]}")
+
+        # Resolve payment gate requirement
+        print("\n=== Step 2: Build & Attach Proofs ===")
+        proofs = []
+        for req in unmet:
+            if req["type"] == "payment":
+                print("Fulfilling payment requirement (Mock USDC token)...")
+                proofs.append({
+                    "type": "payment",
+                    "token": "tok_demo_usdc_0.01"
+                })
+
+        payload["gate_proofs"] = proofs
+
+        # Retry call
+        print("\n=== Step 3: Retry Call with Gate Proofs ===")
+        r = client.post("/mcp/tools/call", json=payload)
+        print(f"Status: {r.status_code}")
+        if r.status_code == 200:
+            result = r.json()["result"]
+            print("Access Granted!")
+            print(f"Result: {json.dumps(result, indent=2, ensure_ascii=False)}...\n")
+        else:
+            print(f"Failed: {r.text}")
+
+    elif r.status_code == 200:
+        print(f"Response from Edgar API:\n{r.json()}")
+
+
+if __name__ == "__main__":
+    main()

examples/edgar-eep-mcp-python/eep_provider.py

@@ -0,0 +1,131 @@
+"""
+SEC Edgar EEP-MCP Provider.
+Exposes FastMCP SEC Tools as gated EEP endpoints.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from eep_mcp_bridge.bridge import run_bridge_server
+from mcp_server import mcp
+import asyncio
+
+tools_from_mcp = asyncio.run(mcp.list_tools())
+
+class FastMCPHttpAdapter(BaseHTTPRequestHandler):
+    """Wraps FastMCP server instance and exposes standard HTTP endpoints."""
+
+    def do_GET(self) -> None:
+        if self.path == "/tools/list":
+            tools_list = []
+            for tool in tools_from_mcp:
+                schema = {}
+                if hasattr(tool, "input_model") and tool.input_model:
+                    if hasattr(tool.input_model, "model_json_schema"):
+                        schema = tool.input_model.model_json_schema()
+                    elif hasattr(tool.input_model, "schema"):
+                        schema = tool.input_model.schema()
+
+                tools_list.append({
+                    "name": tool.name,
+                    "description": tool.description or "",
+                    "inputSchema": schema
+                })
+
+            body = {
+                "server": {"name": mcp.name, "version": "1.0.0"},
+                "tools": tools_list
+            }
+        elif self.path == "/resources/list":
+            body = {"resources": []}
+        else:
+            self.send_response(404)
+            self.end_headers()
+            return
+
+        raw = json.dumps(body).encode("utf-8")
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def do_POST(self) -> None:
+        if self.path != "/tools/call":
+            self.send_response(404)
+            self.end_headers()
+            return
+
+        length = int(self.headers.get("Content-Length", "0"))
+        payload = json.loads(self.rfile.read(length).decode("utf-8")) if length else {}
+        tool_name = payload.get("name")
+        arguments = payload.get("arguments") or {}
+
+        tool = next((t for t in tools_from_mcp if t.name == tool_name), None)
+        if not tool:
+            self.send_response(404)
+            self.end_headers()
+            return
+
+        try:
+            result = tool.fn(**arguments)
+            body = {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]}
+
+            raw = json.dumps(body).encode("utf-8")
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+        except Exception as e:
+            body = {"isError": True, "content": [{"type": "text", "text": str(e)}]}
+            raw = json.dumps(body).encode("utf-8")
+            self.send_response(500)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+    def log_message(self, _format, *_args) -> None:
+        return
+
+
+def start_adapter(port: int = 8001) -> HTTPServer:
+    server = HTTPServer(("127.0.0.1", port), FastMCPHttpAdapter)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    print(f"FastMCP HTTP Adapter running on http://localhost:{port}")
+    return server
+
+
+def main() -> None:
+    adapter_port = 8001
+    adapter = start_adapter(adapter_port)
+
+    config = {
+        "did": "did:web:sec-edgar.eep-dev.org",
+        "base_url": "http://localhost:3005",
+        "mcp_base_url": f"http://localhost:{adapter_port}",
+
+        "gated_tools": {
+            "get_company_facts_tool": {
+                "type": "public"
+            }
+        }
+    }
+
+    try:
+        run_bridge_server(config, host="0.0.0.0", port=3005)
+    except KeyboardInterrupt:
+        print("\nStopping servers...")
+        adapter.shutdown()
+        adapter.server_close()
+
+
+if __name__ == "__main__":
+    main()

examples/edgar-eep-mcp-python/mcp_server.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+from fastmcp import FastMCP
+from pydantic import Field
+from tools.edgar import get_company_facts
+
+
+mcp = FastMCP("SEC Edgar Server")
+
+@mcp.tool()
+def get_company_facts_tool(ticker: str, user_agent:str=Field(description="Given user agent in the format as <project_name> <acme@corp.org>")) -> dict:
+    """Get SEC filing facts (latest assets, net income) for a given company ticker."""
+    return get_company_facts(ticker=ticker,user_agent=user_agent)
+
+if __name__ == "__main__":
+    mcp.run()

examples/edgar-eep-mcp-python/requirements.txt

@@ -0,0 +1,3 @@
+fastmcp>=0.2.0
+httpx>=0.27.0
+requests>=2.31.0

examples/edgar-eep-mcp-python/tools/edgar.py

@@ -0,0 +1,65 @@
+import requests
+from pydantic import Field
+
+
+def get_company_facts(ticker: str, user_agent:str=Field(description="Given user agent in the format as <project_name> <acme@corp.org>")) -> dict:
+    """
+    Fetch basic company data/facts from the SEC EDGAR API.
+    Uses standard SEC requirements (User-Agent header).
+    """
+    ticker_upper = ticker.upper().strip()
+
+    headers = {
+        "User-Agent": user_agent
+    }
+
+    try:
+        mapping_url = "https://www.sec.gov/files/company_tickers.json"
+        res = requests.get(mapping_url, headers=headers, timeout=10)
+        res.raise_for_status()
+        tickers_data = res.json()
+
+        cik = None
+        for item in tickers_data.values():
+            if item["ticker"] == ticker_upper:
+                cik = str(item["cik_str"]).zfill(10)
+                break
+
+        if not cik:
+            return {"error": f"Ticker {ticker_upper} not found in SEC database."}
+
+        facts_url = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{cik}.json"
+        facts_res = requests.get(facts_url, headers=headers, timeout=10)
+        facts_res.raise_for_status()
+        facts = facts_res.json()
+
+        entity_name = facts.get("entityName", "Unknown Company")
+
+        us_gaap = facts.get("facts", {}).get("us-gaap", {})
+
+        assets_data = us_gaap.get("Assets", {}).get("units", {}).get("USD", [])
+        net_income_data = us_gaap.get("NetIncomeLoss", {}).get("units", {}).get("USD", [])
+
+        latest_assets = assets_data[-1] if assets_data else {}
+        latest_net_income = net_income_data[-1] if net_income_data else {}
+
+        return {
+            "ticker": ticker_upper,
+            "company_name": entity_name,
+            "cik": cik,
+            "latest_assets": {
+                "val": latest_assets.get("val"),
+                "fy": latest_assets.get("fy"),
+                "fp": latest_assets.get("fp"),
+                "form": latest_assets.get("form")
+            } if latest_assets else "No data",
+            "latest_net_income": {
+                "val": latest_net_income.get("val"),
+                "fy": latest_net_income.get("fy"),
+                "fp": latest_net_income.get("fp"),
+                "form": latest_net_income.get("form")
+            } if latest_net_income else "No data"
+        }
+
+    except Exception as e:
+        return {"error": f"Failed to retrieve SEC facts: {str(e)}"}

examples/fundturkey-eep-mcp-python/client_demo.py

@@ -13,16 +13,16 @@
 
 def main() -> None:
     client = httpx.Client(base_url=BRIDGE_URL, timeout=15.0)
-    print("🔗 Connecting to FundTurkey EEP-MCP Bridge...")
+    print("Connecting to FundTurkey EEP-MCP Bridge...")
 
     # 1. Manifest Discovery
     r = client.get("/.well-known/eep.json")
-    print(f"✅ Discovered Manifest! Publisher DID: {r.json()['did']}")
+    print(f"Discovered Manifest! Publisher DID: {r.json()['did']}")
 
     # 2. Service Catalog Introspection
     r = client.get("/eep/services")
     tools = [s["id"] for s in r.json()["services"]]
-    print(f"📋 Introspected {len(tools)} tools: {', '.join(tools)}\n")
+    print(f"Introspected {len(tools)} tools: {', '.join(tools)}\n")
 
     # 3. Try to call the gated Premium Comparison tool
     payload = {
@@ -33,22 +33,22 @@ def main() -> None:
         }
     }
 
-    print("━━━ Step 1: Attempt to Call Premium Tool ━━━")
+    print("=== Step 1: Attempt to Call Premium Tool ===")
     r = client.post("/mcp/tools/call", json=payload)
     print(f"Status: {r.status_code}")
 
     if r.status_code == 402:
         error_body = r.json()
-        print(f"❌ Access Restricted! Required Tier: {error_body['required_tier']}")
+        print(f"Access Restricted! Required Tier: {error_body['required_tier']}")
         unmet = error_body["unmet_requirements"]
         print(f"   Unmet requirements: {[u['type'] for u in unmet]}")
 
         # Resolve payment gate requirement
-        print("\n━━━ Step 2: Build & Attach Proofs ━━━")
+        print("\n=== Step 2: Build & Attach Proofs ===")
         proofs = []
         for req in unmet:
             if req["type"] == "payment":
-                print("💳 Fulfilling payment requirement (Mock USDC token)...")
+                print("Fulfilling payment requirement (Mock USDC token)...")
                 proofs.append({
                     "type": "payment",
                     "token": "tok_demo_usdc_0.05"
@@ -57,15 +57,15 @@ def main() -> None:
         payload["gate_proofs"] = proofs
 
         # Retry call
-        print("\n━━━ Step 3: Retry Call with Gate Proofs ━━━")
+        print("\n=== Step 3: Retry Call with Gate Proofs ===")
         r = client.post("/mcp/tools/call", json=payload)
         print(f"Status: {r.status_code}")
         if r.status_code == 200:
             result = r.json()["result"]
-            print("✅ Access Granted!")
-            print(f"📊 Result: {json.dumps(result, indent=2, ensure_ascii=False)}...\n")
+            print("Access Granted!")
+            print(f"Result: {json.dumps(result, indent=2, ensure_ascii=False)}...\n")
         else:
-            print(f"❌ Failed: {r.text}")
+            print(f"Failed: {r.text}")
 
 
 if __name__ == "__main__":