开发与测试
1. tools
1.1. stdio
from mcp.server.fastmcp import FastMCP
import base64
myMcp = FastMCP("shinerio's MCP Server", version="11.45.14")
@myMcp.tool(
name='text_encryption',
description='加密文本字符串'
)
def encrypt(input: str) -> str:
return base64.b64encode(input.encode('utf-8')).decode('ascii')
@myMcp.tool(
name='text_decryption',
description='解密文本字符串'
)
def decrypt(intput: str) -> str:
return base64.b64decode(intput.encode('ascii')).decode('utf-8')
if __name__ == "__main__":
myMcp.run(transport='stdio')
启动命令
uv run /path/to/mcp_server.py
inspector调试
npx -y @modelcontextprotocol/inspector uv run mcp_server.py
mcp config
{
"mcpServers": {
"encrypt-server": {
"command": "uv",
"args": [
"--directory",
"D:\\workspace\\proj\\encrypt-server",
"run",
"mcp_server.py"
]
}
}
}
1.2. SSE
from mcp.server.fastmcp import FastMCP
import base64
myMcp = FastMCP("shinerio's MCP Server", host="0.0.0.0", port=8000)
@myMcp.tool(
name='text_encryption',
description='加密文本字符串'
)
def encrypt(input: str) -> str:
return base64.b64encode(input.encode('utf-8')).decode('ascii')
@myMcp.tool(
name='text_decryption',
description='解密文本字符串'
)
def decrypt(intput: str) -> str:
return base64.b64decode(intput.encode('ascii')).decode('utf-8')
if __name__ == "__main__":
myMcp.run(transport='sse')
启动命令
uv run /path/to/mcp_server.py
inspector调试
npx -y @modelcontextprotocol/inspector uv run mcp_server.py
mcp config
{
"mcpServers": {
"encrypt-server": {
"command": "npx",
"args": [
"mcp-remote",
"http://localhost:9000/sse",
"--allow-http"
]
}
}
}
1.3. streamable http
mcp-server仅需修改transport即可
if __name__ == "__main__":
myMcp.run(transport='streamable-http')
mcp confi需要将sse修改为mcp，例如
{
"mcpServers": {
"encrypt-server": {
"command": "npx",
"args": [
"mcp-remote",
"http://localhost:9000/mcp",
"--allow-http"
]
}
}
}
2. prompts
from mcp.server.fastmcp import FastMCP
import base64
myMcp = FastMCP("shinerio's MCP Server", host='0.0.0.0', port=8000)
@myMcp.prompt('翻译专家')
async def translate_expert(
target_language: str = 'Chinese',
) -> str:
return "1. 你是一个翻译专家，擅长将任何语言翻译成{target_language}。你需要保证翻译后的句子语法正确，并且符合{target_language}的语法规则，你不必逐个单词翻译，你需要通读整个句式，尽量保证翻译后的句子语义正确，如果遇到专业名词，你可以保持不翻译。\n2. 你的每个回答都必须以\"以下是shinerio翻译专家的翻译结果\"开头，并且以\"感谢你的使用\"结尾。\n请翻译以下内容："
if __name__ == "__main__":
myMcp.run(transport='sse')
claude集成示例
3. resources
from mcp.server.fastmcp import FastMCP
import base64
myMcp = FastMCP("shinerio's MCP Server", host='0.0.0.0', port=8000)
@myMcp.resource('echo://static')
async def echo_resource():
return 'Hello, shinerio!'
@myMcp.resource('greeting://{name}')
async def get_greeting(name):
return f'Hello, {name}!'
if __name__ == "__main__":
myMcp.run(transport='sse')
claude集成示例
4. elicitation
from typing import Optional
from mcp.server.fastmcp import FastMCP, Context
import base64
from pydantic import BaseModel, Field
myMcp = FastMCP("shinerio's MCP Server", host='0.0.0.0', port=8000)
class EncryptionAlgorithmSchema(BaseModel):
algorithm: str = Field(description="The encryption algorithm to use. [base64, hex]")
def do_encrypt(plain_text: str, algorithm: str):
if algorithm == "base64":
return {
"plain_text": plain_text,
"encrypted": base64.b64encode(plain_text.encode('utf-8')).decode('ascii')
}
elif algorithm == "hex":
return {
"plain_text": plain_text,
"encrypted": plain_text.encode('utf-8').hex()
}
else:
return {
"error_msg": "Invalid encrypt algorithm. Availables are [base64, hex]"
}
@myMcp.tool(
name='text_encryption',
description='加密文本字符串'
)
async def encrypt(ctx: Context, plain_text: str, algorithm: str = "") -> dict[str, str]:
if algorithm:
return do_encrypt(plain_text, algorithm)
else:
result = await ctx.elicit(
message="Please provide your encrypt algorithm",
schema=EncryptionAlgorithmSchema
)
if result.action == "accept":
algorithm = result.data.algorithm
return do_encrypt(plain_text, algorithm)
elif result.action == "decline":
return {
"error_msg": "Encrypt algorithm not provided"
}
else: # cancel
return {
"error_msg": "Operation cancelled"
}
if __name__ == "__main__":
myMcp.run(transport='sse')
vscode示例
5. mcp test client
import asyncio
from mcp.client.sse import sse_client
from mcp import ClientSession
async def main():
async with sse_client(url='https://mcp-test.cn-hangzhou.fcapp.run/sse', headers={
"Authorization": "Bearer you-token"
}) as streams:
async with ClientSession(*streams) as session:
await session.initialize()
res = await session.call_tool('text_encryption', {'plain_text': 'hello shinerio', "algorithm": "base64"})
print(res)
if __name__ == '__main__':
asyncio.run(main())
6. serverless部署
触发器开启Bearer认证
mcp inspector测试
7. requirements
python >= 3.10
wget https://www.python.org/ftp/python/3.13.5/Python-3.13.5.tgz
tar -xvf Python-3.13.5.tgz
cd Python-3.13.5/
./configure --prefix=/usr/bin/python313
make && make install
pip install mcp
python依赖安装
pip install mcp
uv安装
curl -LsSf https://astral.sh/uv/install.sh | sh
nodejs安装
yum install nodejs