Advanced patterns#
This page explains advanced techniques for building robust MCP servers.
Note
All tool examples use the recommended ctx: Context parameter pattern to access
the application context. For more information, see Function parameter.
Initialize a Python session#
Set up a session with startup code#
You can set up a Python session with custom startup code that runs automatically when the session starts. This approach is useful for importing commonly used libraries, configuring settings, or defining helper functions.
from ansys.common.mcp.helpers import PersistentPythonSession
# Create a session with startup code
startup_code = """
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# Define a helper function
def quick_plot(data):
plt.figure(figsize=(10, 6))
plt.plot(data)
plt.show()
"""
session = PersistentPythonSession(startup_code=startup_code)
session.start()
# Now numpy, pandas, and plt are already imported
result = session.execute("arr = np.array([1, 2, 3, 4, 5])")
When you restart the session using the session.restart() method, the startup
code runs again, ensuring that all imports and configurations are
reestablished. This approach is particularly useful when resetting the session state
while maintaining necessary dependencies.
Run Python code from tools#
The execute_python_code tool lets you run arbitrary Python code in the persistent session.
Because the code runs in the context of the session, it has access to all imports and variables
defined in the startup code.
from mcp.server.fastmcp import Context
from ansys.common.mcp.tools import execute_python_code
@mcp.tool()
async def run_python_code(ctx: Context, code: str) -> str:
"""Run Python code in the persistent session.
Parameters
----------
ctx : Context
MCP context (automatically injected).
code : str
Python code to run.
"""
# Add additional execution logic here (such as logging and error handling)
await return execute_python_code(ctx=ctx, code=code)
Restart a session with history#
You can create a tool to restart the Python session while optionally replaying the command history. This approach allows you to reset the session state without losing previous commands.
@mcp.tool()
def restart_session(ctx: Context, replay_history: bool = True) -> str:
"""Restart the Python session and optionally replay commands.
Parameters
----------
ctx : Context
MCP context (automatically injected).
replay_history : bool, default: True
Whether to replay command history.
"""
app_context = ctx.request_context.lifespan_context
history = app_context.command_history.copy()
result = app_context.python_session.restart()
if not result["success"]:
return f"Restart failed: {result['error']}"
if replay_history and history:
for cmd in history:
app_context.python_session.execute(cmd)
return f"Restarted and replayed {len(history)} commands"
return "Session restarted"
Track command history#
Create and export command history#
You can maintain a command history in the application context and provide tools to export it in various formats.
from mcp.server.fastmcp import Context
@mcp.tool()
def execute_command(ctx: Context, command: str) -> str:
"""Run and track a command.
Parameters
----------
ctx : Context
MCP context (automatically injected).
command : str
Command to run.
"""
app_context = ctx.request_context.lifespan_context
result = app_context.product_instance.run(command)
if result["success"]:
app_context.command_history.append(command)
return result
@mcp.tool()
def export_history(ctx: Context, format: str = "json") -> str:
"""Export the command history as JSON or text.
Parameters
----------
ctx : Context
MCP context (automatically injected).
format : str, default: 'json'
Export format ('json' or 'text').
"""
app_context = ctx.request_context.lifespan_context
if format == "json":
import json
return json.dumps(app_context.command_history, indent=2)
return "\n".join(app_context.command_history)
Handle errors#
Use graceful degradation#
Handle errors without crashing the server:
from ansys.common.mcp.logging_config import get_logger
logger = get_logger(__name__)
def product_startup(self):
"""Start with graceful error handling."""
try:
logger.info("Attempting to connect to product...")
self.context.product_instance = connect(timeout=30)
logger.info(f"Connected: {self.context.product_instance}")
except ConnectionTimeout as e:
logger.error(f"Connection timeout: {e}")
logger.warning("Server will start in limited mode")
self.context.product_instance = None
self.context.metadata["mode"] = "limited"
except Exception as e:
logger.error(f"Unexpected error during startup: {e}")
raise # Re-raise for critical errors
Note
Logs automatically redirect to stderr (not stdout) to avoid interfering with the MCP protocol. The logging configuration handles this behavior.
Add retry logic#
Implement retry logic for flaky connections:
import time
def product_startup(self):
"""Connect with retry logic."""
max_retries = 3
retry_delay = 5 # seconds
for attempt in range(1, max_retries + 1):
try:
logger.info(f"Connection attempt {attempt}/{max_retries}...")
self.context.product_instance = connect()
logger.info("Connected successfully")
return
except Exception as e:
logger.warning(f"Attempt {attempt} failed: {e}")
if attempt < max_retries:
logger.info(f"Retrying in {retry_delay} seconds...")
time.sleep(retry_delay)
else:
logger.error("All connection attempts failed")
raise
Track metadata#
Monitor session state#
from datetime import datetime
import uuid
def product_startup(self):
"""Initialize with state tracking."""
self.context.product_instance = connect()
self.context.metadata.update({
"session_id": str(uuid.uuid4()),
"start_time": datetime.now().isoformat(),
"statistics": {"commands_executed": 0, "errors": 0}
})
Manage user preferences#
@mcp.tool()
def set_preference(ctx: Context, key: str, value: str) -> str:
"""Set a user preference.
Parameters
----------
ctx : Context
MCP context (automatically injected).
key : str
Preference key.
value : str
Preference value.
"""
app_context = ctx.request_context.lifespan_context
app_context.metadata.setdefault("preferences", {})[key] = value
logger.info(f"Set {key} = {value}")
return json.dumps(
{
"success": True,
"stdout": "",
"stderr": "",
"message": "Preference updated",
},
ensure_ascii=False,
indent=2,
)
@mcp.tool()
def get_preference(ctx: Context, key: str, default: str = None) -> str:
"""Get a user preference.
Parameters
----------
ctx : Context
MCP context (automatically injected).
key : str
Preference key.
default : str, default: None
Default value if the specified preference key is not found.
"""
app_context = ctx.request_context.lifespan_context
prefs = app_context.metadata.get("preferences", {})
value = prefs.get(key, default)
if value is None:
return f"Preference '{key}' is not set."
return value
Configure the transport protocol#
By default, MCP servers communicate over stdio, which is the standard transport for local AI client integrations. The base class also supports HTTP transport, which is useful for remote clients, web applications, and CORS-enabled frontends.
The run_cli() method handles all transport
configuration. Call it from your package’s __main__.py instead of
calling app.run() directly:
# src/myproduct_mcp/__main__.py
import sys
from ansys.common.mcp.logging_config import setup_logging
from myproduct_mcp import app
import myproduct_mcp.tools # noqa: F401
def main(argv=None):
setup_logging(level="INFO")
app.run_cli(argv) # transport, host, port, and CORS are handled here
return 0
if __name__ == "__main__":
sys.exit(main())
Your users can then select the transport at startup without any additional code in your package:
# stdio (default)
python -m myproduct_mcp
# HTTP on the default address (127.0.0.1:8080)
python -m myproduct_mcp --transport http
# HTTP with a custom host and port
python -m myproduct_mcp --transport http --http-host 0.0.0.0 --http-port 9000
# HTTP with CORS origins
python -m myproduct_mcp --transport http \\
--cors-origins "http://localhost:3000,https://myapp.com"
Note
HTTP transport uses streamable HTTP (/mcp endpoint). When using the HTTP
transport, start the server process first, then point your AI client at
http://<host>:<port>/mcp.
Add product-specific CLI arguments#
For products that need their own CLI arguments (for example, a connection IP or port),
override _add_cli_arguments() to inject
them into the parser, and
_configure_from_cli() to process them.
The transport dispatch is unchanged — you never need to rewrite it.
Add arguments to the parser:
def _add_cli_arguments(self, parser: argparse.ArgumentParser) -> None:
"""Add PyExample-specific CLI arguments.
Parameters
----------
parser : argparse.ArgumentParser
Argument parser pre-populated with the standard transport arguments.
Add product-specific arguments directly to it.
"""
parser.add_argument(
"--ip",
dest="example_ip",
default="127.0.0.1",
help="PyExample server IP or hostname.",
)
parser.add_argument(
"--port",
dest="example_port",
type=int,
default=50052,
help="PyExample gRPC port.",
)
parser.add_argument(
"--connect-on-startup",
dest="connect_on_startup",
action="store_true",
help="Connect to PyExample during MCP startup.",
)
Store the parsed values before the server starts:
def _configure_from_cli(self, args: argparse.Namespace) -> None:
"""Store parsed PyExample CLI arguments before the server starts.
Parameters
----------
args : argparse.Namespace
Fully parsed namespace containing both standard transport arguments
and the product-specific arguments added by :meth:`_add_cli_arguments`.
"""
self._cli_config = {
"example_ip": args.example_ip,
"example_port": args.example_port,
"connect_on_startup": args.connect_on_startup,
}
The resulting server now accepts both the standard transport arguments and the product-specific ones:
python -m pyexample_mcp --ip 10.0.0.5 --port 50052 --connect-on-startup
python -m pyexample_mcp --transport http --http-port 9000 --ip 10.0.0.5
Test CLI argument parsing#
Because run_cli() accepts an optional argv list, you can test CLI argument
parsing without starting a real server. Tests for transport and CORS:
class TestMainCliDispatch:
"""Tests verifying that CLI arguments reach run_cli() correctly via main()."""
def test_stdio_transport(self):
"""Test that stdio transport args are forwarded."""
with patch.object(app, "run_async", return_value=None) as mock_run:
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--transport", "stdio"])
mock_run.assert_called_once()
def test_http_transport_default_host_and_port(self):
"""Test that HTTP transport uses default host and port when not specified."""
with patch.object(app, "run_http_async", return_value=None) as mock_http:
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--transport", "http"])
mock_http.assert_called_once_with(
transport="http",
host="127.0.0.1",
port=8080,
middleware=None,
)
def test_http_transport_custom_host_and_port(self):
"""Test that custom --http-host and --http-port are forwarded."""
with patch.object(app, "run_http_async", return_value=None) as mock_http:
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--transport", "http", "--http-host", "0.0.0.0", "--http-port", "9000"])
mock_http.assert_called_once_with(
transport="http",
host="0.0.0.0",
port=9000,
middleware=None,
)
def test_http_transport_cors_origins(self):
"""Test that --cors-origins creates a CORSMiddleware passed via middleware."""
from starlette.middleware.cors import CORSMiddleware
with patch.object(app, "run_http_async", return_value=None) as mock_http:
with patch("ansys.common.mcp.server.asyncio.run"):
main(
[
"--transport",
"http",
"--cors-origins",
"http://localhost:3000,https://myapp.com",
]
)
mock_http.assert_called_once()
mw_list = mock_http.call_args.kwargs["middleware"]
assert len(mw_list) == 1
assert mw_list[0].cls is CORSMiddleware
assert mw_list[0].kwargs["allow_origins"] == [
"http://localhost:3000",
"https://myapp.com",
]
def test_invalid_port_causes_system_exit(self):
"""Test that an out-of-range port causes SystemExit."""
with pytest.raises(SystemExit):
main(["--transport", "http", "--http-port", "99999"])
def test_unknown_transport_causes_system_exit(self):
"""Test that an unknown transport value causes SystemExit."""
with pytest.raises(SystemExit):
main(["--transport", "grpc"])
Tests for product-specific arguments:
class TestPyExampleCliHooks:
"""Unit tests for PyExampleMCP._add_cli_arguments and _configure_from_cli."""
def test_add_cli_arguments_registers_ip(self):
"""Test that --ip is registered and stores value in example_ip."""
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--ip", "10.0.0.5"])
assert app._cli_config["example_ip"] == "10.0.0.5"
def test_add_cli_arguments_registers_port(self):
"""Test that --port is registered and stored as an integer."""
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--port", "50099"])
assert app._cli_config["example_port"] == 50099
def test_add_cli_arguments_registers_connect_on_startup_flag(self):
"""Test that --connect-on-startup sets the flag to True."""
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--connect-on-startup"])
assert app._cli_config["connect_on_startup"] is True
def test_configure_from_cli_defaults(self):
"""Test that _configure_from_cli stores default values when no product args given."""
with patch("ansys.common.mcp.server.asyncio.run"):
main([])
assert app._cli_config == {
"example_ip": "127.0.0.1",
"example_port": 50052,
"connect_on_startup": False,
}
def test_configure_from_cli_all_product_args(self):
"""Test that all product-specific args are stored correctly in _cli_config."""
with patch("ansys.common.mcp.server.asyncio.run"):
main(["--ip", "192.168.1.1", "--port", "12345", "--connect-on-startup"])
assert app._cli_config == {
"example_ip": "192.168.1.1",
"example_port": 12345,
"connect_on_startup": True,
}
Expose tool sets#
A tool set groups tools under a named tag.
Important
Registering the toolsets://definition resource is required for integration with
some Ansys products, as it allows the product to discover and display available
tool sets in the user interface.
Use @app.tool(tags={...}) to assign a tool to one or more sets, and
@app.resource("toolsets://definition") to expose the tool set definitions as a list.
Each tool set must include a name, description, skill (instructions for the AI
agent on when and how to use the tools), and tools (list of tool function names).
Register the resource:
@app.resource("toolsets://definition")
def list_tool_sets() -> list[dict]:
"""Toolset definitions."""
return [
{
"name": "structures",
"description": "Tools for creating and running structural models",
"skill": (
"Use these tools to set up and solve structural simulations. "
"Start with create_model to define the model, then use "
"run_simulation to execute the analysis."
),
"tools": ["create_model", "run_simulation"],
},
{
"name": "post_processing",
"description": "Tools for processing and exporting simulation results",
"skill": (
"Use these tools to inspect results and run custom analyses after a simulation. "
"Use get_command_history to review past commands and execute_python_code "
"for custom post-processing scripts."
),
"tools": ["get_command_history", "execute_python_code"],
},
]
Tag a tool with the structures set:
@app.tool(tags={"structures"})
def create_model(
ctx: Context,
name: str,
model_type: str = "default",
parameters: Optional[dict] = None,
) -> str:
"""Create a new model in PyExample.
Parameters
----------
ctx : Context
The FastMCP context
name : str
Name for the new model
model_type : str
Type of model to create
parameters : Optional[dict]
Additional model parameters
Returns
-------
str
Status message
"""
app_context = ctx.fastmcp._lifespan_result
if not app_context.example_instance:
return "Error: PyExample not connected"
# Create model (simulated)
params = parameters or {}
model = app_context.example_instance.create_model(name, model_type, **params)
# Update command history
command = f"CREATE MODEL {name} TYPE {model_type}"
app_context.command_history.append(command)
logger.info(f"Created model: {name} (type: {model_type})")
return f"Model '{name}' created successfully\n{model}"
Tag a tool with the post_processing set:
@app.tool(tags={"post_processing"})
def get_command_history(ctx: Context, format: str = "list") -> str:
"""Retrieve command execution history.
Parameters
----------
ctx : Context
The FastMCP context
format : str
Output format: 'list', 'numbered', or 'json'
Returns
-------
str
Command history in requested format
"""
app_context = ctx.fastmcp._lifespan_result
if not app_context.command_history:
return "No commands executed yet"
if format == "numbered":
lines = [f"{i + 1}. {cmd}" for i, cmd in enumerate(app_context.command_history)]
return "\n".join(lines)
elif format == "json":
return json.dumps(app_context.command_history, indent=2)
else: # list format
return "\n".join(app_context.command_history)
A tool can belong to multiple sets by listing several tags:
@app.tool(tags={"structures", "post_processing"})
def get_stress_report(ctx: Context, model_name: str) -> str:
"""Generate a stress report."""
...