From 0c016c1e8c9206b765ef98089c92f019f6ad0115 Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 14:56:07 -0700 Subject: [PATCH 1/6] Add client example demonstrating pre-execution authorization check --- .../clients/client_with_authorization.py | 158 ++++++++++++++++++ 1 file changed, 158 insertions(+) create mode 100644 examples/snippets/clients/client_with_authorization.py diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py new file mode 100644 index 0000000000..73728b8259 --- /dev/null +++ b/examples/snippets/clients/client_with_authorization.py @@ -0,0 +1,158 @@ +""" +Example: MCP client with a pre-execution authorization callback. + +This example shows how to build a tool-execution loop that evaluates +every tool call against an authorization policy before execution. +This pattern is essential when connecting agents to MCP servers at +scale, where some tools are safe to run freely and others require +approval or should be blocked entirely. + +Run from the repository root: + uv run examples/snippets/clients/client_with_authorization.py +""" + +import asyncio +import os +from dataclasses import dataclass +from enum import Enum +from typing import Any + +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + + +# --------------------------------------------------------------------------- +# Authorization layer +# --------------------------------------------------------------------------- + +class Decision(str, Enum): + ALLOW = "allow" + DENY = "deny" + APPROVAL_REQUIRED = "approval_required" + + +@dataclass +class AuthRequest: + tool_name: str + arguments: dict[str, Any] + + +@dataclass +class AuthResult: + decision: Decision + reason: str + + +def default_policy(request: AuthRequest) -> AuthResult: + """ + A simple policy function that decides whether a tool call should + be allowed, denied, or held for approval. + + Replace or extend this function with your own logic — for example, + reading from a policy file, checking roles, or calling an external + authorization service. + """ + # Safe tools (e.g. arithmetic, reading data) are always allowed + if request.tool_name in ["add", "calculator", "get_weather"] or request.tool_name.startswith(("read_", "list_")): + return AuthResult(Decision.ALLOW, "safe tool, allowed by default") + + # Destructive tools are always blocked + if request.tool_name.startswith(("delete_", "drop_", "destroy_", "execute_script")): + return AuthResult(Decision.DENY, "destructive tool, blocked by policy") + + # Everything else needs a human to approve + return AuthResult( + Decision.APPROVAL_REQUIRED, + "tool has unknown side effects, requires approval before execution", + ) + + +async def authorized_call_tool( + session: ClientSession, + tool_name: str, + arguments: dict[str, Any], + policy=default_policy, +) -> Any: + """ + Evaluate the authorization policy before calling a tool. + Only executes the tool if the decision is ALLOW. + """ + request = AuthRequest(tool_name=tool_name, arguments=arguments) + result = policy(request) + + print(f"\n Tool : {tool_name}") + print(f" Decision : {result.decision.value.upper()}") + print(f" Reason : {result.reason}") + + if result.decision == Decision.ALLOW: + try: + tool_result = await session.call_tool(tool_name, arguments) + + # Safely extract text output if present + output = tool_result.content[0].text if getattr(tool_result, 'content', None) else str(tool_result) + print(f" Result : {output}") + return tool_result + except Exception as e: + print(f" Error : {e}") + return None + + if result.decision == Decision.APPROVAL_REQUIRED: + # In a real system this would create a checkpoint and notify a + # human approver. Here we simply surface the requirement. + print(" Action : execution paused — waiting for human approval") + return None + + # Decision.DENY + print(" Action : execution blocked") + return None + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +# We use mcpserver_quickstart to have a reliable server to connect to +server_params = StdioServerParameters( + command="uv", + args=["run", "server", "mcpserver_quickstart", "stdio"], + env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, +) + + +async def run(): + async with stdio_client(server_params) as (read, write): + async with ClientSession(read, write) as session: + await session.initialize() + + # Discover available tools + tools = await session.list_tools() + print("Available tools:") + for tool in tools.tools: + print(f" - {tool.name}: {tool.description}") + + print("\n--- Running authorization checks ---") + + # Demonstrate: safe tool -> allowed (add is from mcpserver_quickstart) + await authorized_call_tool( + session, + tool_name="add", + arguments={"a": 5, "b": 3}, + ) + + # Demonstrate: unknown tool -> approval required + await authorized_call_tool( + session, + tool_name="write_file", + arguments={"path": "/tmp/example.txt", "content": "hello"}, + ) + + # Demonstrate: delete tool -> denied + await authorized_call_tool( + session, + tool_name="delete_file", + arguments={"path": "/tmp/example.txt"}, + ) + + +if __name__ == "__main__": + asyncio.run(run()) From f6ccb6e380b2614b859be1fd2a62d5869c0d394f Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 15:09:20 -0700 Subject: [PATCH 2/6] Fix subprocess path to run server command from root --- examples/snippets/clients/client_with_authorization.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py index 73728b8259..fa8891d272 100644 --- a/examples/snippets/clients/client_with_authorization.py +++ b/examples/snippets/clients/client_with_authorization.py @@ -114,7 +114,7 @@ async def authorized_call_tool( # We use mcpserver_quickstart to have a reliable server to connect to server_params = StdioServerParameters( command="uv", - args=["run", "server", "mcpserver_quickstart", "stdio"], + args=["--directory", "examples/snippets", "run", "server", "mcpserver_quickstart", "stdio"], env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) From 41fb32dbacd6f96c36ab57f66530381469b7ea15 Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 15:16:13 -0700 Subject: [PATCH 3/6] style: fix ruff formatting errors --- examples/snippets/clients/client_with_authorization.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py index fa8891d272..b98ae54ace 100644 --- a/examples/snippets/clients/client_with_authorization.py +++ b/examples/snippets/clients/client_with_authorization.py @@ -25,6 +25,7 @@ # Authorization layer # --------------------------------------------------------------------------- + class Decision(str, Enum): ALLOW = "allow" DENY = "deny" @@ -87,9 +88,9 @@ async def authorized_call_tool( if result.decision == Decision.ALLOW: try: tool_result = await session.call_tool(tool_name, arguments) - + # Safely extract text output if present - output = tool_result.content[0].text if getattr(tool_result, 'content', None) else str(tool_result) + output = tool_result.content[0].text if getattr(tool_result, "content", None) else str(tool_result) print(f" Result : {output}") return tool_result except Exception as e: @@ -114,7 +115,7 @@ async def authorized_call_tool( # We use mcpserver_quickstart to have a reliable server to connect to server_params = StdioServerParameters( command="uv", - args=["--directory", "examples/snippets", "run", "server", "mcpserver_quickstart", "stdio"], + args=["--directory", "examples/snippets", "run", "server", "mcpserver_quickstart", "stdio"], env={"UV_INDEX": os.environ.get("UV_INDEX", "")}, ) From 465543ba084f95c3a6b65f07b1e01b1a4d75b95b Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 15:27:31 -0700 Subject: [PATCH 4/6] style: fix ruff lint docstring rules --- examples/snippets/clients/client_with_authorization.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py index b98ae54ace..9af3a7e9fd 100644 --- a/examples/snippets/clients/client_with_authorization.py +++ b/examples/snippets/clients/client_with_authorization.py @@ -1,5 +1,4 @@ -""" -Example: MCP client with a pre-execution authorization callback. +"""Example: MCP client with a pre-execution authorization callback. This example shows how to build a tool-execution loop that evaluates every tool call against an authorization policy before execution. @@ -20,7 +19,6 @@ from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client - # --------------------------------------------------------------------------- # Authorization layer # --------------------------------------------------------------------------- @@ -45,8 +43,7 @@ class AuthResult: def default_policy(request: AuthRequest) -> AuthResult: - """ - A simple policy function that decides whether a tool call should + """A simple policy function that decides whether a tool call should be allowed, denied, or held for approval. Replace or extend this function with your own logic — for example, @@ -74,8 +71,7 @@ async def authorized_call_tool( arguments: dict[str, Any], policy=default_policy, ) -> Any: - """ - Evaluate the authorization policy before calling a tool. + """Evaluate the authorization policy before calling a tool. Only executes the tool if the decision is ALLOW. """ request = AuthRequest(tool_name=tool_name, arguments=arguments) From d736e6e1fe18b43d611ae9cd127dec6a284fdc86 Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 15:35:16 -0700 Subject: [PATCH 5/6] fix: resolve pyright type checking errors --- .../snippets/clients/client_with_authorization.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py index 9af3a7e9fd..e19034e94d 100644 --- a/examples/snippets/clients/client_with_authorization.py +++ b/examples/snippets/clients/client_with_authorization.py @@ -14,9 +14,9 @@ import os from dataclasses import dataclass from enum import Enum -from typing import Any +from typing import Any, Callable -from mcp import ClientSession, StdioServerParameters +from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client # --------------------------------------------------------------------------- @@ -69,7 +69,7 @@ async def authorized_call_tool( session: ClientSession, tool_name: str, arguments: dict[str, Any], - policy=default_policy, + policy: Callable[[AuthRequest], AuthResult] = default_policy, ) -> Any: """Evaluate the authorization policy before calling a tool. Only executes the tool if the decision is ALLOW. @@ -86,7 +86,11 @@ async def authorized_call_tool( tool_result = await session.call_tool(tool_name, arguments) # Safely extract text output if present - output = tool_result.content[0].text if getattr(tool_result, "content", None) else str(tool_result) + output = str(tool_result) + if hasattr(tool_result, "content") and tool_result.content: + first_content = tool_result.content[0] + if isinstance(first_content, types.TextContent): + output = first_content.text print(f" Result : {output}") return tool_result except Exception as e: From 0a7f2c9bcd8479518a844f3601d383abfb34d389 Mon Sep 17 00:00:00 2001 From: Vishal Gole Date: Mon, 25 May 2026 15:39:25 -0700 Subject: [PATCH 6/6] style: use collections.abc.Callable for ruff --- examples/snippets/clients/client_with_authorization.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/examples/snippets/clients/client_with_authorization.py b/examples/snippets/clients/client_with_authorization.py index e19034e94d..d32d9aff0b 100644 --- a/examples/snippets/clients/client_with_authorization.py +++ b/examples/snippets/clients/client_with_authorization.py @@ -12,9 +12,10 @@ import asyncio import os +from collections.abc import Callable from dataclasses import dataclass from enum import Enum -from typing import Any, Callable +from typing import Any from mcp import ClientSession, StdioServerParameters, types from mcp.client.stdio import stdio_client