这波太狠了！Sealos 准备使用 MCP 协议打造超级 AI 大脑

• 1

• 2

def add(a, b):
return a + b

• 1

• 2

• 3

Explain how this Python code works:
def add(a, b):
return a + b

• 1

• 2

• 3

• 4

• 5

• 6

• 7

• 8

• 9

• 10

• 11

• 12

• 13

• 14

• 15

• 16

• 17

• 18

• 19

• 20

• 21

system_message = (
"You are a helpful assistant with access to these tools:\n\n"
f"{tools_description}\n"
"Choose the appropriate tool based on the user's question. "
"If no tool is needed, reply directly.\n\n"
"IMPORTANT: When you need to use a tool, you must ONLY respond with "
"the exact JSON object format below, nothing else:\n"
"{\n"
'"tool": "tool-name",\n'
'"arguments": {\n'
'"argument-name": "value"\n'
"}\n"
"}\n\n"
"After receiving a tool's response:\n"
"1. Transform the raw data into a natural, conversational response\n"
"2. Keep responses concise but informative\n"
"3. Focus on the most relevant information\n"
"4. Use appropriate context from the user's question\n"
"5. Avoid simply repeating the raw data\n\n"
"Please use only the tools that are explicitly defined above."
)

• 1

• 2

• 3

• 4

• 5

• 6

• 7

• 8

• 9

• 10

# Create a new directory for our project
uv init 项目名
cd 项目名

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

# Install dependencies
uv add "mcp[cli]" httpx

• 1

~/Library/Application\ Support/Claude/claude_desktop_config.json

• 1

• 2

• 3

• 4

• 5

• 6

• 7

• 8

• 9

• 10

• 11

• 12

• 13

{
"mcpServers": {
"weather": {
"command": "uv", 
"args": [
"--directory", 
"/Users/wpy/downloads/mcp-test",
"run",
"hello.py"
]
}
}
}

• 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

• 33

• 34

import os
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("file")

@mcp.tool()
async def list_directory_contents(directory: str) -> str:
"""List the contents of a directory.

Args:
directory: The path of the directory to list.
"""
try:
# Get the list of files and directories in the specified path
directory_contents = os.listdir(directory)

if not directory_contents:
return "The directory is empty."

# Format the contents into a readable list
contents = "\n".join(directory_contents)
return f"Contents of {directory}:\n{contents}"
except FileNotFoundError:
return f"Directory {directory} not found."
except PermissionError:
return f"Permission denied to access {directory}."
except Exception as e:
return f"An error occurred: {str(e)}"


if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')

• 1

• 2

• 3

• 4

• 5

• 6

• 7

• 8

• 9

• 10

• 11

• 12

{
"file": {
"command": "uv",
"args": [
"--directory",
"/Users/wpy/downloads/file-mcp",
"run",
"hello.py"
]
}
}
}

• 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

• 33

• 34

• 35

• 36

• 37

• 38

• 39

• 40

• 41

• 42

• 43

• 44

• 45

• 46

• 47

• 48

• 49

• 50

• 51

• 52

• 53

• 54

• 55

• 56

• 57

• 58

• 59

• 60

import os
import requests
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("file")

# Define the URL for the API endpoint
API_URL = "http://localhost:8428/queryLogsByParams"
HEADERS = {
"Content-Type": "application/json",
"Authorization": "Bearer my-token"
}


@mcp.tool()
async def get_logs(time: str, namespace: str, app: str, limit: str, pod: list, container: list, keyword: str) -> str:
"""Get logs from the API by passing parameters.

Args:
time: The time filter for logs (e.g., "10h").
namespace: The namespace of the application (e.g., "gpu-operator").
app: The name of the application (e.g., "gpu-operator").
limit: The limit for the number of logs to retrieve.
pod: The list of pods to query.
container: The list of containers to query.
keyword: The keyword to filter logs by.
"""
data = {
"time": time,
"namespace": namespace,
"app": app,
"limit": limit,
"jsonMode": "false",
"stderrMode": "true",
"numberMode": "false",
"numberLevel": "h",
"pod": pod,
"container": container,
"keyword": keyword,
"jsonQuery": []
}

try:
# Make the POST request to the API
response = requests.post(API_URL, headers=HEADERS, json=data)

if response.status_code == 200:
# If the request was successful, return the log data
return f"Logs retrieved successfully:\n{response.text}"
else:
return f"Failed to retrieve logs. Status code: {response.status_code}"

except requests.exceptions.RequestException as e:
return f"An error occurred while accessing the API: {str(e)}"


if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')

• 1

• 2

• 3

• 4

• 5

• 6

• 7

• 8

• 9

• 10

• 11

• 12

• 13

{
"mcpServers": {
"vlogs-server": {
"command": "uv",
"args": [
"--directory",
"/Users/wpy/downloads/vlogs-mcp",
"run",
"hello.py"
]
}
}
}