from fastcore.utils import *
from inspect import Parameter, Signature
from typing import Any
from toolslm.funccall import schema2sig,mk_tool
import httpx, jsonfrom pprint import pprinturl = 'https://mcp.grep.app'headers = {"Accept": "application/json, text/event-stream"}
c = httpx.Client(base_url=url, headers=headers)def xget (path='/', *args, **kwargs): return c.get (path, *args, **kwargs)
def xpost(path='/', *args, **kwargs): return c.post(path, *args, **kwargs)@patch
def rpc(self:httpx.Client, path='/', *args, params=None, jsonrpc="2.0", id_=None, **kwargs):
kwargs['jsonrpc'] = jsonrpc
kwargs['id'] = id_ or id(kwargs)
kwargs['params'] = params or {}
return self.post(path, *args, json=kwargs)init_params = dict( protocolVersion="2024-11-05", capabilities={},
clientInfo={"name": "manual-client", "version": "1.0"})
r = c.rpc(method="initialize", params=init_params)
r.status_code200print(r.text)event: message
data: {"result":{"protocolVersion":"2024-11-05","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"mcp-typescript server on vercel","version":"0.1.0"}},"jsonrpc":"2.0","id":5166467008}
r = c.rpc(method="tools/list")
print(r.text)event: message
data: {"result":{"tools":[{"name":"searchGitHub","description":"Find real-world code examples from over a million public GitHub repositories to help answer programming questions.\n\n**IMPORTANT: This tool searches for literal code patterns (like grep), not keywords. Search for actual code that would appear in files:**\n- ✅ Good: 'useState(', 'import React from', 'async function', '(?s)try {.*await'\n- ❌ Bad: 'react tutorial', 'best practices', 'how to use'\n\n**When to use this tool:**\n- When implementing unfamiliar APIs or libraries and need to see real usage patterns\n- When unsure about correct syntax, parameters, or configuration for a specific library\n- When looking for production-ready examples and best practices for implementation\n- When needing to understand how different libraries or frameworks work together\n\n**Perfect for questions like:**\n- \"How do developers handle authentication in Next.js apps?\" → Search: 'getServerSession' with language=['TypeScript', 'TSX']\n- \"What are common React error boundary patterns?\" → Search: 'ErrorBoundary' with language=['TSX']\n- \"Show me real useEffect cleanup examples\" → Search: '(?s)useEffect\\(\\(\\) => {.*removeEventListener' with useRegexp=true\n- \"How do developers handle CORS in Flask applications?\" → Search: 'CORS(' with matchCase=true and language=['Python']\n\nUse regular expressions with useRegexp=true for flexible patterns like '(?s)useState\\(.*loading' to find useState hooks with loading-related variables. Prefix the pattern with '(?s)' to match across multiple lines.\n\nFilter by language, repository, or file path to narrow results.","inputSchema":{"type":"object","properties":{"query":{"type":"string","description":"The literal code pattern to search for (e.g., 'useState(', 'export function'). Use actual code that would appear in files, not keywords or questions."},"matchCase":{"type":"boolean","description":"Whether the search should be case sensitive","default":false},"matchWholeWords":{"type":"boolean","description":"Whether to match whole words only","default":false},"useRegexp":{"type":"boolean","description":"Whether to interpret the query as a regular expression","default":false},"repo":{"type":"string","description":"Filter by repository.\n Examples: 'facebook/react', 'microsoft/vscode', 'vercel/ai'.\n Can match partial names, for example 'vercel/' will find repositories in the vercel org."},"path":{"type":"string","description":"Filter by file path.\n Examples: 'src/components/Button.tsx', 'README.md'.\n Can match partial paths, for example '/route.ts' will find route.ts files at any level."},"language":{"type":"array","items":{"type":"string"},"description":"Filter by programming language.\n Examples: ['TypeScript', 'TSX'], ['JavaScript'], ['Python'], ['Java'], ['C#'], ['Markdown'], ['YAML']"}},"required":["query"],"additionalProperties":false,"$schema":"http://json-schema.org/draft-07/schema#"}}]},"jsonrpc":"2.0","id":5148335040}
def parse_sse(text):
events, ev = [], {}
for line in text.split('\n'):
if not line:
if ev: events.append(ev); ev = {}
elif not line.startswith(':') and ':' in line:
k, _, v = line.partition(':')
if v.startswith(' '): v = v[1:]
ev[k] = f"{ev[k]}\n{v}" if k in ev else v
return dict2obj(events + ([ev] if ev else []))# parse_sse(r.text)def parse_mcp(r):
ct = r.headers.get('content-type', '')
if ct.startswith('text/event-stream'):
events = parse_sse(r.text)
return [dict2obj(json.loads(e['data'])) for e in events if 'data' in e]
elif ct.startswith('application/json'): return [r.json()]
else: raise ValueError(f"Unknown content-type: {ct}")# parse_mcp(r)sr = parse_mcp(r)[0]
list(sr)['result', 'jsonrpc', 'id']list(sr.result)['tools']tools = sr.result.tools
len(tools)1tool = sr.result.tools[0]def shorten_desc(d):
if isinstance(d, dict):
return {k: (v[:10] + '…' if k == 'description' and isinstance(v, str) else shorten_desc(v)) for k, v in d.items()}
if isinstance(d, list): return [shorten_desc(x) for x in d]
return dbrf_tool = shorten_desc(tool)
pprint(brf_tool, width=120, indent=2){ 'description': 'Find real-…',
'inputSchema': { '$schema': 'http://json-schema.org/draft-07/schema#',
'additionalProperties': False,
'properties': { 'language': { 'description': 'Filter by …',
'items': {'type': 'string'},
'type': 'array'},
'matchCase': {'default': False, 'description': 'Whether th…', 'type': 'boolean'},
'matchWholeWords': { 'default': False,
'description': 'Whether to…',
'type': 'boolean'},
'path': {'description': 'Filter by …', 'type': 'string'},
'query': {'description': 'The litera…', 'type': 'string'},
'repo': {'description': 'Filter by …', 'type': 'string'},
'useRegexp': {'default': False, 'description': 'Whether to…', 'type': 'boolean'}},
'required': ['query'],
'type': 'object'},
'name': 'searchGitHub'}@patch
def method(self:httpx.Client, name=None, method='tools/call', **kwargs):
params = dict(arguments=kwargs)
if name: params['name'] = name
return parse_mcp(self.rpc(method=method, params=params))r = c.method(name="searchGitHub", query='class PtyProcess', language=['Python'])resps = r[0].result.content
len(resps)10resps[0]{ 'text': 'Repository: sedwards2009/extraterm\n'
'Path: '
'extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py\n'
'URL: '
'https://github.com/sedwards2009/extraterm/blob/master/extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py\n'
'License: MIT\n'
'\n'
'Snippets:\n'
'--- Snippet 1 (Line 79) ---\n'
' (intr, eof) = (3, 4)\n'
' \n'
' _INTR = _byte(intr)\n'
' _EOF = _byte(eof)\n'
'\n'
'class PtyProcessError(Exception):\n'
' """Generic error class for this package."""\n'
'\n'
'# setecho and setwinsize are pulled out here because on some '
'platforms, we need\n'
'# to do this from the child before we exec()\n'
' \n'
'\n'
'\n'
'--- Snippet 2 (Line 119) ---\n'
" TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)\n"
' # Note, assume ws_xpixel and ws_ypixel are zero.\n'
" s = struct.pack('HHHH', rows, cols, 0, 0)\n"
' fcntl.ioctl(fd, TIOCSWINSZ, s)\n'
'\n'
'class PtyProcess(object):\n'
" '''This class represents a process running in a "
'pseudoterminal.\n'
' \n'
' The main constructor is the :meth:`spawn` classmethod.\n',
'type': 'text'}print(resps[0].text)Repository: sedwards2009/extraterm
Path: extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py
URL: https://github.com/sedwards2009/extraterm/blob/master/extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py
License: MIT
Snippets:
--- Snippet 1 (Line 79) ---
(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
class PtyProcessError(Exception):
"""Generic error class for this package."""
# setecho and setwinsize are pulled out here because on some platforms, we need
# to do this from the child before we exec()
--- Snippet 2 (Line 119) ---
TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)
# Note, assume ws_xpixel and ws_ypixel are zero.
s = struct.pack('HHHH', rows, cols, 0, 0)
fcntl.ioctl(fd, TIOCSWINSZ, s)
class PtyProcess(object):
'''This class represents a process running in a pseudoterminal.
The main constructor is the :meth:`spawn` classmethod.
def parse_result(text):
header, _, snips = text.partition('Snippets:\n')
result = {k.lower(): v for line in header.strip().split('\n') if ': ' in line for k, v in [line.split(': ', 1)]}
parts = re.split(r'--- Snippet \d+ \(Line (\d+)\) ---\n', snips)[1:]
return dict2obj(result | {'snippets':
[{'line': int(ln), 'code': code.strip()} for ln, code in zip(parts[::2], parts[1::2])]})res = parse_result(resps[0].text)res{ 'license': 'MIT',
'path': 'extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py',
'repository': 'sedwards2009/extraterm',
'snippets': [{'line': 79, 'code': '(intr, eof) = (3, 4)\n \n _INTR = _byte(intr)\n _EOF = _byte(eof)\n\nclass PtyProcessError(Exception):\n """Generic error class for this package."""\n\n# setecho and setwinsize are pulled out here because on some platforms, we need\n# to do this from the child before we exec()'}, {'line': 119, 'code': "TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)\n # Note, assume ws_xpixel and ws_ypixel are zero.\n s = struct.pack('HHHH', rows, cols, 0, 0)\n fcntl.ioctl(fd, TIOCSWINSZ, s)\n\nclass PtyProcess(object):\n '''This class represents a process running in a pseudoterminal.\n \n The main constructor is the :meth:`spawn` classmethod."}],
'url': 'https://github.com/sedwards2009/extraterm/blob/master/extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py'}schema2sig(tool)<Signature (query: Any, *, matchCase: Any = False, matchWholeWords: Any = False, useRegexp: Any = False, repo: Any = None, path: Any = None, language: list[typing.Any] = None)>class MCPException(Exception): pass
class MCPClient:
def __init__(self, url):
self._c = httpx.Client(base_url=url, headers={"Accept": "application/json, text/event-stream"})
self.tools = dict2obj({t.name: t for t in self._call(method="tools/list").tools})
def _call(self, *args, **kwargs):
resp = self._c.method(*args, **kwargs)[0]
if 'error' in resp: raise MCPException(resp.error.message)
return resp.result
def __call__(self, name, **kwargs): return self._call(name, **kwargs)
def __dir__(self): return list(self.tools.keys()) + list(super().__dir__())
def __getattr__(self, name):
if name not in self.tools: raise AttributeError(name)
return mk_tool(self, self.tools[name])mcp = MCPClient("https://mcp.grep.app")
list(mcp.tools)['searchGitHub']mcp.searchGitHub?Signature: mcp.searchGitHub( query: Any, *, matchCase: Any = False, matchWholeWords: Any = False, useRegexp: Any = False, repo: Any = None, path: Any = None, language: list[typing.Any] = None, ) Docstring: Find real-world code examples from over a million public GitHub repositories to help answer programming questions. **IMPORTANT: This tool searches for literal code patterns (like grep), not keywords. Search for actual code that would appear in files:** - ✅ Good: 'useState(', 'import React from', 'async function', '(?s)try {.*await' - ❌ Bad: 'react tutorial', 'best practices', 'how to use' **When to use this tool:** - When implementing unfamiliar APIs or libraries and need to see real usage patterns - When unsure about correct syntax, parameters, or configuration for a specific library - When looking for production-ready examples and best practices for implementation - When needing to understand how different libraries or frameworks work together **Perfect for questions like:** - "How do developers handle authentication in Next.js apps?" → Search: 'getServerSession' with language=['TypeScript', 'TSX'] - "What are common React error boundary patterns?" → Search: 'ErrorBoundary' with language=['TSX'] - "Show me real useEffect cleanup examples" → Search: '(?s)useEffect\(\(\) => {.*removeEventListener' with useRegexp=true - "How do developers handle CORS in Flask applications?" → Search: 'CORS(' with matchCase=true and language=['Python'] Use regular expressions with useRegexp=true for flexible patterns like '(?s)useState\(.*loading' to find useState hooks with loading-related variables. Prefix the pattern with '(?s)' to match across multiple lines. Filter by language, repository, or file path to narrow results. File: ~/aai-ws/toolslm/toolslm/funccall.py Type: function
r = mcp.searchGitHub('class PtyProcess', language=['Python'])
list(r)['content']c = r.content[0]
list(c)['type', 'text']pr = parse_result(c.text)
pr{ 'license': 'MIT',
'path': 'extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py',
'repository': 'sedwards2009/extraterm',
'snippets': [{'line': 79, 'code': '(intr, eof) = (3, 4)\n \n _INTR = _byte(intr)\n _EOF = _byte(eof)\n\nclass PtyProcessError(Exception):\n """Generic error class for this package."""\n\n# setecho and setwinsize are pulled out here because on some platforms, we need\n# to do this from the child before we exec()'}, {'line': 119, 'code': "TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)\n # Note, assume ws_xpixel and ws_ypixel are zero.\n s = struct.pack('HHHH', rows, cols, 0, 0)\n fcntl.ioctl(fd, TIOCSWINSZ, s)\n\nclass PtyProcess(object):\n '''This class represents a process running in a pseudoterminal.\n \n The main constructor is the :meth:`spawn` classmethod."}],
'url': 'https://github.com/sedwards2009/extraterm/blob/master/extensions/ProxySessionBackend/src/python/ptyprocess/ptyprocess.py'}for o in pr.snippets: print(o.code, '\n---\n')(intr, eof) = (3, 4)
_INTR = _byte(intr)
_EOF = _byte(eof)
class PtyProcessError(Exception):
"""Generic error class for this package."""
# setecho and setwinsize are pulled out here because on some platforms, we need
# to do this from the child before we exec()
---
TIOCSWINSZ = getattr(termios, 'TIOCSWINSZ', -2146929561)
# Note, assume ws_xpixel and ws_ypixel are zero.
s = struct.pack('HHHH', rows, cols, 0, 0)
fcntl.ioctl(fd, TIOCSWINSZ, s)
class PtyProcess(object):
'''This class represents a process running in a pseudoterminal.
The main constructor is the :meth:`spawn` classmethod.
---
See if you can find some recent info on most free online popular MCPs. I would love to find 1-2 that use streamable http (without SSE) to test that part of our code. Let me know what you find.
🤖Reply🤖
🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠🧠
Good news! I found some useful info on free online MCPs, including ones that support Streamable HTTP:
Free Public MCPs with Streamable HTTP:
* * DeepWiki / GitMCP - Free, remote, no-authentication-required service. Server URLs:
https://mcp.deepwiki.com/sse(SSE) orhttps://mcp.deepwiki.com/mcp(Streamable HTTP). Offers instant MCP server creation for any GitHub repository.* Semgrep - Uses the popular open-source security scanner to scan code for vulnerabilities. A streamable-http endpoint is hosted at
https://mcp.semgrep.ai/mcp.
About Streamable HTTP vs SSE:
* Streamable HTTP is an HTTP-based communication mechanism that supports streaming responses and bidirectional interaction on a single HTTP connection. The MCP client sends requests to the server via HTTP POST, and the server can respond either with a normal single JSON message or by initiating a live event stream (using Server-Sent Events, SSE) to stream back multiple messages over time.
* Connections start as standard HTTP requests, but can dynamically upgrade to SSE to stream responses during long-running tasks. When an AI agent wants to call a tool on a remote MCP server, it can do so with a single POST request to one endpoint (/mcp). Depending on the tool call, the server will either respond immediately or decide to upgrade the connection to use SSE.
Key insight for testing: The Streamable HTTP transport can return plain JSON responses without SSE for simple operations. * To detect which transport a server supports: attempt to POST an InitializeRequest to the server URL - if it succeeds, the client can assume this is a server supporting the new Streamable HTTP transport.
So to test your code with pure HTTP (no SSE), you could try: - https://mcp.deepwiki.com/mcp - https://mcp.semgrep.ai/mcp
Want me to try hitting one of these to see if they return plain JSON vs SSE?
url = "https://mcp.deepwiki.com/mcp"
mcp = MCPClient(url)
list(mcp.tools)--------------------------------------------------------------------------- ValueError Traceback (most recent call last) Cell In[1], line 2 1 url = "https://mcp.deepwiki.com/mcp" ----> 2 mcp = MCPClient(url) 3 list(mcp.tools) Cell In[1], line 6, in MCPClient.__init__(self, url) 4 def __init__(self, url): 5 self._c = httpx.Client(base_url=url, headers={"Accept": "application/json, text/event-stream"}) ----> 6 self.tools = dict2obj({t.name: t for t in self._call(method="tools/list").tools}) Cell In[1], line 9, in MCPClient._call(self, *args, **kwargs) 8 def _call(self, *args, **kwargs): ----> 9 resp = self._c.method(*args, **kwargs)[0] 10 if 'error' in resp: raise MCPException(resp.error.message) 11 return resp.result Cell In[1], line 5, in method(self, name, method, **kwargs) 3 params = dict(arguments=kwargs) 4 if name: params['name'] = name ----> 5 return parse_mcp(self.rpc(method=method, params=params)) Cell In[1], line 7, in parse_mcp(r) 5 return [dict2obj(json.loads(e['data'])) for e in events if 'data' in e] 6 elif ct.startswith('application/json'): return [r.json()] ----> 7 else: raise ValueError(f"Unknown content-type: {ct}") ValueError: Unknown content-type: text/plain;charset=UTF-8
Let’s work through debugging this together to get it all working.
DeepWiki MCP
How to use the official DeepWiki MCP server
The DeepWiki MCP server provides programmatic access to DeepWiki’s public repository documentation and search capabilities (Ask Devin).
[
](#what-is-mcp)
What is MCP?
The Model Context Protocol (MCP) is an open standard that enables AI apps to securely connect to MCP-compatible data sources and tools. You can think of MCP like a USB-C port for AI applications - a standardized way to connect AI apps to different services.
[
](#deepwiki-mcp-server)
DeepWiki MCP Server
The DeepWiki MCP server is a free, remote, no-authentication-required service that provides access to public repositories. Base Server URL: https://mcp.deepwiki.com/
[
](#available-tools)
Available Tools
The DeepWiki MCP server offers three main tools:
read_wiki_structure- Get a list of documentation topics for a GitHub repositoryread_wiki_contents- View documentation about a GitHub repositoryask_question- Ask any question about a GitHub repository and get an AI-powered, context-grounded response
[
](#wire-protocols)
Wire Protocols
The DeepWiki MCP server supports two wire protocols:
[
](#sse-server-sent-events-/sse)
SSE (Server-Sent Events) - /sse
- URL:
https://mcp.deepwiki.com/sse - The official MCP specification version
- Supported by Claude
- Recommended for most integrations
[
](#streamable-http-/mcp)
Streamable HTTP - /mcp
- URL:
https://mcp.deepwiki.com/mcp - Newer protocol, works with Cloudflare and OpenAI
- Also supports the legacy
/sseversion
For maximum compatibility, try the SSE endpoint at /sse first.
[
](#setup-instructions)
Setup Instructions
[
](#for-most-clients-e-g-windsurf,-cursor-:)
For most clients (e.g. Windsurf, Cursor):
Copy
Ask AI
{
"mcpServers": {
"deepwiki": {
"serverUrl": "https://mcp.deepwiki.com/sse"
}
}
}[
](#for-claude-code:)
For Claude Code:
Copy
Ask AI
claude mcp add -s user -t http deepwiki https://mcp.deepwiki.com/mcp[
](#related-resources)