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