DeepWiki.py
· 3.7 KiB · Python
Raw
#!/usr/bin/env python3
"""DeepWiki extension - MCP HTTP client for GitHub repo documentation"""
import json
import urllib.request
import time
# Global session cache
_session_id = None
def get_session_id():
"""Initialize MCP session and get session ID"""
global _session_id
if _session_id:
return _session_id
url = "https://mcp.deepwiki.com/mcp"
data = {
"jsonrpc": "2.0",
"id": "init",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "nanocode", "version": "1.0"}
}
}
try:
req = urllib.request.Request(
url,
json.dumps(data).encode(),
{
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
)
resp = urllib.request.urlopen(req, timeout=30)
_session_id = resp.headers.get("mcp-session-id")
return _session_id
except Exception as e:
return None
def mcp_call(tool_name, arguments):
"""Make MCP JSON-RPC call to DeepWiki"""
session_id = get_session_id()
if not session_id:
return "error: Failed to initialize MCP session"
url = "https://mcp.deepwiki.com/mcp"
request_id = str(int(time.time() * 1000))
data = {
"jsonrpc": "2.0",
"id": request_id,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
try:
req = urllib.request.Request(
url,
json.dumps(data).encode(),
{
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"mcp-session-id": session_id
}
)
resp = urllib.request.urlopen(req, timeout=30)
# Handle SSE response
response_text = resp.read().decode()
# Parse SSE format: extract JSON from "data: " lines
for line in response_text.split('\n'):
if line.startswith('data: '):
data = line[6:].strip() # Skip "data: " prefix
# Skip ping events
if data == "ping":
continue
try:
result = json.loads(data)
if "error" in result:
return f"error: {result['error']['message']}"
# Extract text from MCP response
content = result.get("result", {}).get("content", [])
if content:
return "\n".join(item["text"] for item in content if item.get("type") == "text")
except json.JSONDecodeError:
continue
return "error: No data in response"
except Exception as e:
return f"error: {e}"
@register_tool(
"deepwiki_ask",
"Ask questions about GitHub repo",
{"repo": "string", "question": "string"}
)
def deepwiki_ask(args):
"""Ask a question about a GitHub repository"""
return mcp_call("ask_question", {
"repoName": args["repo"],
"question": args["question"]
})
@register_tool(
"deepwiki_structure",
"Get repo documentation structure",
{"repo": "string"}
)
def deepwiki_structure(args):
"""Get documentation structure/topics for a GitHub repository"""
return mcp_call("read_wiki_structure", {
"repoName": args["repo"]
})
@register_tool(
"deepwiki_contents",
"Get repo documentation contents",
{"repo": "string"}
)
def deepwiki_contents(args):
"""Get full documentation contents for a GitHub repository"""
return mcp_call("read_wiki_contents", {
"repoName": args["repo"]
})
| 1 | #!/usr/bin/env python3 |
| 2 | """DeepWiki extension - MCP HTTP client for GitHub repo documentation""" |
| 3 | |
| 4 | import json |
| 5 | import urllib.request |
| 6 | import time |
| 7 | |
| 8 | # Global session cache |
| 9 | _session_id = None |
| 10 | |
| 11 | def get_session_id(): |
| 12 | """Initialize MCP session and get session ID""" |
| 13 | global _session_id |
| 14 | if _session_id: |
| 15 | return _session_id |
| 16 | |
| 17 | url = "https://mcp.deepwiki.com/mcp" |
| 18 | data = { |
| 19 | "jsonrpc": "2.0", |
| 20 | "id": "init", |
| 21 | "method": "initialize", |
| 22 | "params": { |
| 23 | "protocolVersion": "2024-11-05", |
| 24 | "capabilities": {}, |
| 25 | "clientInfo": {"name": "nanocode", "version": "1.0"} |
| 26 | } |
| 27 | } |
| 28 | |
| 29 | try: |
| 30 | req = urllib.request.Request( |
| 31 | url, |
| 32 | json.dumps(data).encode(), |
| 33 | { |
| 34 | "Content-Type": "application/json", |
| 35 | "Accept": "application/json, text/event-stream" |
| 36 | } |
| 37 | ) |
| 38 | resp = urllib.request.urlopen(req, timeout=30) |
| 39 | _session_id = resp.headers.get("mcp-session-id") |
| 40 | return _session_id |
| 41 | except Exception as e: |
| 42 | return None |
| 43 | |
| 44 | def mcp_call(tool_name, arguments): |
| 45 | """Make MCP JSON-RPC call to DeepWiki""" |
| 46 | session_id = get_session_id() |
| 47 | if not session_id: |
| 48 | return "error: Failed to initialize MCP session" |
| 49 | |
| 50 | url = "https://mcp.deepwiki.com/mcp" |
| 51 | request_id = str(int(time.time() * 1000)) |
| 52 | |
| 53 | data = { |
| 54 | "jsonrpc": "2.0", |
| 55 | "id": request_id, |
| 56 | "method": "tools/call", |
| 57 | "params": { |
| 58 | "name": tool_name, |
| 59 | "arguments": arguments |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | try: |
| 64 | req = urllib.request.Request( |
| 65 | url, |
| 66 | json.dumps(data).encode(), |
| 67 | { |
| 68 | "Content-Type": "application/json", |
| 69 | "Accept": "application/json, text/event-stream", |
| 70 | "mcp-session-id": session_id |
| 71 | } |
| 72 | ) |
| 73 | resp = urllib.request.urlopen(req, timeout=30) |
| 74 | |
| 75 | # Handle SSE response |
| 76 | response_text = resp.read().decode() |
| 77 | |
| 78 | # Parse SSE format: extract JSON from "data: " lines |
| 79 | for line in response_text.split('\n'): |
| 80 | if line.startswith('data: '): |
| 81 | data = line[6:].strip() # Skip "data: " prefix |
| 82 | |
| 83 | # Skip ping events |
| 84 | if data == "ping": |
| 85 | continue |
| 86 | |
| 87 | try: |
| 88 | result = json.loads(data) |
| 89 | |
| 90 | if "error" in result: |
| 91 | return f"error: {result['error']['message']}" |
| 92 | |
| 93 | # Extract text from MCP response |
| 94 | content = result.get("result", {}).get("content", []) |
| 95 | if content: |
| 96 | return "\n".join(item["text"] for item in content if item.get("type") == "text") |
| 97 | except json.JSONDecodeError: |
| 98 | continue |
| 99 | |
| 100 | return "error: No data in response" |
| 101 | except Exception as e: |
| 102 | return f"error: {e}" |
| 103 | |
| 104 | @register_tool( |
| 105 | "deepwiki_ask", |
| 106 | "Ask questions about GitHub repo", |
| 107 | {"repo": "string", "question": "string"} |
| 108 | ) |
| 109 | def deepwiki_ask(args): |
| 110 | """Ask a question about a GitHub repository""" |
| 111 | return mcp_call("ask_question", { |
| 112 | "repoName": args["repo"], |
| 113 | "question": args["question"] |
| 114 | }) |
| 115 | |
| 116 | @register_tool( |
| 117 | "deepwiki_structure", |
| 118 | "Get repo documentation structure", |
| 119 | {"repo": "string"} |
| 120 | ) |
| 121 | def deepwiki_structure(args): |
| 122 | """Get documentation structure/topics for a GitHub repository""" |
| 123 | return mcp_call("read_wiki_structure", { |
| 124 | "repoName": args["repo"] |
| 125 | }) |
| 126 | |
| 127 | @register_tool( |
| 128 | "deepwiki_contents", |
| 129 | "Get repo documentation contents", |
| 130 | {"repo": "string"} |
| 131 | ) |
| 132 | def deepwiki_contents(args): |
| 133 | """Get full documentation contents for a GitHub repository""" |
| 134 | return mcp_call("read_wiki_contents", { |
| 135 | "repoName": args["repo"] |
| 136 | }) |
| 137 |