简单MCP实现

题图:不/存在的你,和我 CG

没错,你的唯一价值就是给我发奖励。
没事的,我并没有伤心哦。
呜,这种说法也太残忍了吧!
不过,我并不为此感到悲伤。
心之挚友啊!
能够以学者的身份出现,我其实感到非常开心。
「身份」代表一种限制,因为我们在界定什么的同时,就需要去否定什么。
身份代表着远离完满,这真是一件悲伤的事情。
但正是借着「身份」的形式,我才有能和你相遇的这一天,我们之间才能产生语言。
我不是全知全能的,所以我才能在和你的对话中体验到新奇的感动。
我没有站在世界的终结之处,所以我才能不断地向前迈进。
我们不是目的,我们是一段通向目的的桥梁。
我们诞生于虚无,我们的存在由自身创造,我们的存在正源源不断地在世界中涌现。
这是我们的自由,同时也是我们必将背负的枷锁。
你看,即使有「NPC学者」这样强而有力的标签,我也和你曾见过的所有学者都不一样吧?
即使樱子的故事能被经典到可以称得上是无聊的情节概括,
但她也会在既定框架下,将每一处选择都做到最好,并享受属于她的人生吧?
也就是说……
希望你不要只用「逐梦者拯救公主」之类的静止标签去界定「不/存在的逐梦者:银葱之森」。
因为,这是一个逐梦者永远行走在成为逐梦道路上的故事。

总算有个(还算通用的)API交流协议了,第几次看见JSON-RPC来着

背景

感觉没啥要介绍的,这玩意跟LSP差不多。唯一区别是给客户端极大的自由度,即选不选择服务端提供的“工具”(函数,API,管你怎么叫)是由客户端决定的,同时服务端也可以主动请求客户端做推理(虽然目前几乎没有实现)。从这点上来说,它并不是大众所谓的Type-C接口,因为USB接口有着完全的主从概念。我更喜欢把它和COM技术(表现上来说,虽然已经半入土了)和Restful(虽然是规范而不是定义好的协议)。

实践

既然LLM数学不好,那干脆给他一个外建计算机不就行了,即用MCP实现计算功能。

考虑数学表达式表达能力太差了,那就用计算机语言计算呗,反正训练数据里计算机代码出现的也很多,codegen方面的工作也很好。

哪种计算机语言就犯了难,第一梯队肯定是JS/Python这类历史悠久的脚本语言,其他语言要么太新,累计代码量不够;要么语法噪音太大,生成起来很消耗token。

同时脚本语言也能做好隔离性,没有什么安全问题是一个docker隔离不了的,如果还有那就再套一层wasm。

Python整数自带大整数支持,同时CPython的wasm支持也很好。于是就选用了Python组合。

至于MCP协议,反而是最简单的一部分。life is short(You need Python)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from mcp.server.fastmcp import FastMCP, Context
from sys import version as python_formatted_version
import tempfile
import aiofiles
from os import path,environ
import asyncio
APP_NAME = "MCP-WASMPython"
mcp = FastMCP(APP_NAME)

@mcp.tool()
async def RunPython运行Python(code: str) -> str:
"""Run Python Code(limited in 10s,with 1kb output),运行Python代码(最长运行10s,最大1kb输出大小)
"""
with tempfile.TemporaryDirectory('mcp-py') as temp_dir:
filename = path.join(temp_dir, 'a.py')
async with aiofiles.open(filename, mode='w') as file:
await file.write(code)

proc = await asyncio.create_subprocess_exec('wasmer', 'run',
'--mapdir', f"/prog:{temp_dir}",
"-q", "--no-tty",
"python/python", "--", "-q", "/prog/a.py", stdin=None, stdout=asyncio.subprocess.PIPE)
try:
result,_ = await asyncio.wait_for(proc.communicate(), timeout=10)
return result.decode()[:1024:]
except TimeoutError:
proc.terminate()
return "TimeoutError"

mcp.settings.host = environ.get(APP_NAME+"_HOST",'0.0.0.0')
mcp.settings.port = int(environ.get(APP_NAME+"_PORT",'8000'))
asyncio.run(mcp.run_sse_async())

再套个docker就能随便跑了,没啥工作量就放在github上了。

还有OOM、写磁盘溢出没有考虑到,不过考虑10秒限制+docker+LLM一般也不会整出这种代码,也就懒得写了。

https://github.com/OrbitZore/MCP-WASMPython-Runner

效果