Showing 1–30 of 79 findings

high (1)

medium (24)

low (5)

«‹

79 findings

high (22)

AIVSS 6.0CVSS 8.1

Command injection risk. Shell-execution sink called with interpolated / attacker-controllable input. Use list-arg subprocess with shell=False, or escape every variable via shlex.quote (Python) / shell-escape (Node).

Evidence

# Try both npx.cmd and npx.exe on Windows
        for cmd in ["npx.cmd", "npx.exe", "npx"]:
            try:
                subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=True)
                return cmd
            except subprocess.CalledProcessError:
                continue

Remediation

Replace shell-based execution with list-arg subprocess calls (shell=False) or escape every interpolated value with shlex.quote (Python) / shell-escape (Node). Treat every MCP tool input as hostile.

45	# Try both npx.cmd and npx.exe on Windows
46	for cmd in ["npx.cmd", "npx.exe", "npx"]:
47	try:
48	subprocess.run([cmd, "--version"], check=True, capture_output=True, shell=True)
49	return cmd
50	except subprocess.CalledProcessError:
51	continue

1	# MCP Python SDK
2
3	<div align="center">
4
5	<strong>Python implementation of the Model Context Protocol (MCP)</strong>
6
7	[![PyPI][pypi-badge]][pypi-url]
8	[![MIT licensed][mit-badge]][mit-url]
9	[![Python Version][python-badge]][python-url]
10	[![Documentation][docs-badge]][docs-url]
11	[![Protocol][protocol-badge]][protocol-url]
12	[![Specification][spec-badge]][spec-url]
13
14	</div>
15
16	<!-- TODO(v2): Replace this README with README.v2.md when v2 is released -->
17
18	> [!NOTE]
19	> **This README documents v1.x of the MCP Pyt

1	from mcp.server.mcpserver import MCPServer
2	from mcp.server.mcpserver.prompts import base
3
4	mcp = MCPServer(name="Prompt Example")
5
6
7	@mcp.prompt(title="Code Review")
8	def review_code(code: str) -> str:
9	return f"Please review this code:\n\n{code}"
10
11
12	@mcp.prompt(title="Debug Assistant")
13	def debug_error(error: str) -> list[base.Message]:
14	return [
15	base.UserMessage("I'm seeing this error:"),
16	base.UserMessage(error),
17	base.AssistantMessage("I'll help debug that. What have y

1	"""MCPServer Echo Server"""
2
3	from mcp.server.mcpserver import MCPServer
4
5	# Create server
6	mcp = MCPServer("Echo Server")
7
8
9	@mcp.tool()
10	def echo_tool(text: str) -> str:
11	"""Echo the input text"""
12	return text
13
14
15	@mcp.resource("echo://static")
16	def echo_resource() -> str:
17	return "Echo!"
18
19
20	@mcp.resource("echo://{text}")
21	def echo_template(text: str) -> str:
22	"""Echo the input text"""
23	return f"Echo: {text}"
24
25
26	@mcp.prompt("echo")
27	def echo_prompt(text: str) -> str:
28	return text

1	"""MCPServer quickstart example.
2
3	Run from the repository root:
4	uv run examples/snippets/servers/mcpserver_quickstart.py
5	"""
6
7	from mcp.server.mcpserver import MCPServer
8
9	# Create an MCP server
10	mcp = MCPServer("Demo")
11
12
13	# Add an addition tool
14	@mcp.tool()
15	def add(a: int, b: int) -> int:
16	"""Add two numbers"""
17	return a + b
18
19
20	# Add a dynamic greeting resource
21	@mcp.resource("greeting://{name}")
22	def get_greeting(name: str) -> str:
23	"""Get a personalized greeting"""
24	return f"Hello, {n

1	from mcp.server.mcpserver import MCPServer
2	from mcp.types import (
3	Completion,
4	CompletionArgument,
5	CompletionContext,
6	PromptReference,
7	ResourceTemplateReference,
8	)
9
10	mcp = MCPServer(name="Example")
11
12
13	@mcp.resource("github://repos/{owner}/{repo}")
14	def github_repo(owner: str, repo: str) -> str:
15	"""GitHub repository resource."""
16	return f"Repository: {owner}/{repo}"
17
18
19	@mcp.prompt(description="Code review prompt")
20	def review_code(language: str, code: str) -> str:
21	"""Gen

1	# MCP Python SDK
2
3	!!! info "You are viewing the in-development v2 documentation"
4	For the current stable release, see the [v1.x documentation](https://py.sdk.modelcontextprotocol.io/).
5
6	The Model Context Protocol (MCP) allows applications to provide context for LLMs in a standardized way, separating the concerns of providing context from the actual LLM interaction.
7
8	This Python SDK implements the full MCP specification, making it easy to:
9
10	- Build MCP servers that expose resources, pr

1	from __future__ import annotations
2
3	from datetime import datetime
4	from typing import Annotated, Any, Final, Generic, Literal, TypeAlias, TypeVar
5
6	from pydantic import BaseModel, ConfigDict, Field, FileUrl, TypeAdapter
7	from pydantic.alias_generators import to_camel
8	from typing_extensions import NotRequired, TypedDict
9
10	from mcp.types.jsonrpc import RequestId
11
12	LATEST_PROTOCOL_VERSION = "2025-11-25"
13	"""The latest version of the Model Context Protocol.
14
15	You can find the latest specification at https:

1	"""MCP Server Module
2
3	This module provides a framework for creating an MCP (Model Context Protocol) server.
4	It allows you to easily define and handle various types of requests and notifications
5	using constructor-based handler registration.
6
7	Usage:
8	1. Define handler functions:
9	async def my_list_tools(ctx, params):
10	return types.ListToolsResult(tools=[...])
11
12	async def my_call_tool(ctx, params):
13	return types.CallToolResult(content=[...])
14
15	2. Create a Server instance with on_* han

1	"""Run from the repository root:
2	uv run examples/snippets/servers/streamable_config.py
3	"""
4
5	from mcp.server.mcpserver import MCPServer
6
7	mcp = MCPServer("StatelessServer")
8
9
10	# Add a simple tool to demonstrate the server
11	@mcp.tool()
12	def greet(name: str = "World") -> str:
13	"""Greet someone by name."""
14	return f"Hello, {name}!"
15
16
17	# Run server with streamable_http transport
18	# Transport-specific options (stateless_http, json_response) are passed to run()
19	if __name__ == "__main__":
20	# Stateles

1	#!/usr/bin/env python3
2	"""MCP Everything Server - Conformance Test Server
3
4	Server implementing all MCP features for conformance testing based on Conformance Server Specification.
5	"""
6
7	import asyncio
8	import base64
9	import json
10	import logging
11
12	import click
13	from mcp.server import ServerRequestContext
14	from mcp.server.mcpserver import Context, MCPServer
15	from mcp.server.mcpserver.prompts.base import UserMessage
16	from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
17	from mcp.type

1	#!/usr/bin/env python3
2	"""MCP Everything Server - Conformance Test Server
3
4	Server implementing all MCP features for conformance testing based on Conformance Server Specification.
5	"""
6
7	import asyncio
8	import base64
9	import json
10	import logging
11
12	import click
13	from mcp.server import ServerRequestContext
14	from mcp.server.mcpserver import Context, MCPServer
15	from mcp.server.mcpserver.prompts.base import UserMessage
16	from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
17	from mcp.type

1	# Migration Guide: v1 to v2
2
3	This guide covers the breaking changes introduced in v2 of the MCP Python SDK and how to update your code.
4
5	## Overview
6
7	Version 2 of the MCP Python SDK introduces several breaking changes to improve the API, align with the MCP specification, and provide better type safety.
8
9	## Breaking Changes
10
11	### `streamablehttp_client` removed
12
13	The deprecated `streamablehttp_client` function has been removed. Use `streamable_http_client` instead.
14
15	Before (v1):
16
17	```python
18	from

370	# Or for SSE
371	mcp = MCPServer("Server")
372	mcp.run(transport="sse", host="0.0.0.0", port=9000, sse_path="/events")
373	```
374
375	For mounted apps:

1	"""Run from the repository root:
2	uv run examples/snippets/servers/oauth_server.py
3	"""
4
5	from pydantic import AnyHttpUrl
6
7	from mcp.server.auth.provider import AccessToken, TokenVerifier
8	from mcp.server.auth.settings import AuthSettings
9	from mcp.server.mcpserver import MCPServer
10
11
12	class SimpleTokenVerifier(TokenVerifier):
13	"""Simple token verifier for demonstration."""
14
15	async def verify_token(self, token: str) -> AccessToken \| None:
16	pass # This is where you would implement actual to

355	mcp.run(transport="streamable-http")
356
357	# Or for SSE
358	mcp = FastMCP("Server", host="0.0.0.0", port=9000, sse_path="/events")
359	mcp.run(transport="sse")
360	```

74	message = params.message
75
76	if not url:
77	print("Error: No URL provided in elicitation request")
78	return types.ElicitResult(action="cancel")
79
80	# Reject dangerous URL schemes before prompting the user

896	async def handle_progress(ctx: ServerRequestContext, params: ProgressNotificationParams) -> None:
897	print(f"Progress: {params.progress}/{params.total}")
898
899	server = Server("my-server", on_progress=handle_progress)
900	```

100	async def handle_elicitation(context, params: ElicitRequestParams) -> ElicitResult:
101	# Display the message to the user
102	print(f"Server asks: {params.message}")
103
104	# Collect user input (this is a simplified example)
105	response = input("Your response (y/n): ")

2115	async def handle_sampling_message(
2116	context: ClientRequestContext, params: types.CreateMessageRequestParams
2117	) -> types.CreateMessageResult:
2118	print(f"Sampling request: {params.messages}")
2119	return types.CreateMessageResult(
2120	role="assistant",
2121	content=types.TextContent(

2175	async def handle_sampling_message(
2176	context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
2177	) -> types.CreateMessageResult:
2178	print(f"Sampling request: {params.messages}")
2179	return types.CreateMessageResult(
2180	role="assistant",
2181	content=types.TextContent(

287	async def elicitation_callback(context, params: ElicitRequestParams) -> ElicitResult:
288	print(f"\n[Elicitation] {params.message}")
289	response = input("Confirm? (y/n): ")
290	return ElicitResult(action="accept", content={"confirm": response.lower() == "y"})

342	def main() -> None:
343	"""Main entry point for the conformance client."""
344	if len(sys.argv) < 2:
345	print(f"Usage: {sys.argv[0]} <server-url>", file=sys.stderr)
346	sys.exit(1)
347
348	server_url = sys.argv[1]

27	params: ElicitRequestParams,
28	) -> ElicitResult:
29	"""Handle elicitation requests from the server."""
30	print(f"\n[Elicitation] Server asks: {params.message}")
31
32	# Simple terminal prompt
33	response = input("Your response (y/n): ").strip().lower()

github · python-sdke8e64842781c

LLM summary— python-sdk:e8e64842781c

Signal scores

LLM judge panel

CVE check

Community trust

Scan details

Tool heatmap

MCP Server Information

Run this exact engine on your private repo — before users see it on /scan.

high (1)

medium (24)

low (5)

high (22)

medium (52)

low (5)

Tool heatmap

Severity breakdown

243	return f"Elicitation result: action={result.action}, content={content}"
244	except Exception as e:
245	return f"Elicitation not supported or error: {str(e)}"
246
247
248	class EnumSchemasTestSchema(BaseModel):

189	return f"LLM response: {model_response}"
190	except Exception as e:
191	return f"Sampling not supported or error: {str(e)}"
192
193
194	class UserResponse(BaseModel):

211	return f"User response: action={result.action}, content={content}"
212	except Exception as e:
213	return f"Elicitation not supported or error: {str(e)}"
214
215
216	class SEP1034DefaultsSchema(BaseModel):

302	return f"Elicitation completed: action={result.action}, content={content}"
303	except Exception as e:
304	return f"Elicitation not supported or error: {str(e)}"
305
306
307	@mcp.tool()

313	except MCPError:
314	raise
315	except Exception as e:
316	return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)
317	if isinstance(result, CallToolResult):
318	return result
319	if isinstance(result, tuple) and len(result) == 2:

206	async def update_importance(user_embedding: list[float], deps: Deps):
207	async with deps.pool.acquire() as conn:
208	rows = await conn.fetch("SELECT id, importance, access_count, embedding FROM memories")
209	for row in rows:
210	memory_embedding = row["embedding"]
211	similarity = cosine_similarity(user_embedding, memory_embedding)

1	# /// script
2	# dependencies = ["pydantic-ai-slim[openai]", "asyncpg", "numpy", "pgvector"]
3	# ///
4
5	# uv pip install 'pydantic-ai-slim[openai]' asyncpg numpy pgvector
6
7	"""Recursive memory system inspired by the human brain's clustering of memories.
8	Uses OpenAI's 'text-embedding-3-small' model and pgvector for efficient
9	similarity search.
10	"""
11
12	import asyncio
13	import math
14	import os
15	from dataclasses import dataclass
16	from datetime import datetime, timezone
17	from pathlib import Path
18	from typing import An

1722	description="Query the database",
1723	inputSchema={
1724	"type": "object",
1725	"properties": {"query": {"type": "string", "description": "SQL query to execute"}},
1726	"required": ["query"],
1727	},
1728	)

29	# see OAuthClientMetadata; we only support `code`
30	response_type: Literal["code"] = Field(..., description="Must be 'code' for authorization code flow")
31	code_challenge: str = Field(..., description="PKCE code challenge")
32	code_challenge_method: Literal["S256"] = Field("S256", description="PKCE code challenge method, must be S256")
33	state: str \| None = Field(None, description="Optional state parameter")
34	scope: str \| None = Field(

33	"""Parameters for initializing an sse_client."""
34
35	# The endpoint URL.
36	url: str
37
38	# Optional headers to include in requests.
39	headers: dict[str, Any] \| None = None