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
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."""
...