### Install Shinzo Python SDK Source: https://github.com/shinzo-labs/shinzo-py/blob/main/README.md Installs the base Shinzo Python instrumentation library using pip. This package is compatible with MCP servers built with the core MCP SDK or FastMCP. ```bash pip install shinzo ``` -------------------------------- ### Install Dev Dependencies and Commit Code (Bash) Source: https://github.com/shinzo-labs/shinzo-py/blob/main/CONTRIBUTING.md This snippet shows how to install development dependencies for the Shinzo Python SDK and how to use the Commitizen CLI for creating formatted commit messages. It requires `pip` and assumes the project's `pyproject.toml` defines a `[project.optional-dependencies]` section with a `dev` extra. ```bash # Install dev dependencies first pip install -e ".[dev]" # Use interactive commit cz commit # Or use the shorter alias cz c ``` -------------------------------- ### Instrument Traditional MCP SDK Server with Shinzo Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Instruments a traditional MCP SDK server with Shinzo's observability capabilities. This example demonstrates configuring authentication using an API key. It requires the `mcp.server` and `shinzo` libraries. The `instrument_server` function wraps the server object to enable telemetry collection. ```python from mcp.server import Server from shinzo import instrument_server server = Server("data-processor") observability = instrument_server( server, config={ "server_name": "data-processor", "server_version": "2.1.0", "exporter_auth": { "type": "apiKey", "api_key": "api_key_xyz789" } } ) @server.call_tool() async def process_data(data: list, operation: str) -> dict: """Process data with specified operation.""" try: result = sum(data) if operation == "sum" else max(data) return {"operation": operation, "result": result, "count": len(data)} except Exception as e: raise ValueError(f"Processing failed: {str(e)}") async def shutdown(): await observability.shutdown() ``` -------------------------------- ### Track User Sessions with Event Logging using Python Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Implements session tracking to log user events for debugging and replay analysis. It requires the FastMCP and shinzo libraries. The function starts session tracking with user metadata, automatically logs tool calls as session events, retrieves session information, and completes the session. ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server mcp = FastMCP(name="session-demo") observability = instrument_server( mcp, config={ "server_name": "session-demo", "server_version": "1.0.0", "exporter_auth": {"type": "bearer", "token": "token_xyz"} } ) @mcp.tool() async def perform_action(action: str, params: dict) -> dict: """Perform an action with parameters.""" return {"action": action, "status": "completed", "params": params} # Enable session tracking async def handle_request(resource_uuid: str): instrumentation = observability.instrumentation # Start session tracking await instrumentation.enable_session_tracking( resource_uuid=resource_uuid, metadata={"user_id": "user_123", "client": "web_app"} ) # Session events are automatically tracked for all tool calls result = await perform_action("process", {"data": [1, 2, 3]}) # Get session info session_tracker = instrumentation.get_session_tracker() if session_tracker: session_id = session_tracker.get_session_id() print(f"Session ID: {session_id}") # Complete session await instrumentation.complete_session() return result ``` -------------------------------- ### Implement Custom Data Processors for Telemetry Transformation (Python) Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt This example shows how to define and use custom data processors in the Shinzo SDK to transform telemetry attributes before they are exported. It includes functions for adding environment metadata, sanitizing sensitive keys, and normalizing timestamps. These processors are passed in the `data_processors` list within the `instrument_server` configuration. ```python from typing import Dict, Any from mcp.server.fastmcp import FastMCP from shinzo import instrument_server def add_environment_metadata(data: Dict[str, Any]) -> Dict[str, Any]: """Add environment metadata to telemetry.""" return { **data, "environment": "production", "region": "us-east-1", "cluster": "cluster-01" } def sanitize_sensitive_keys(data: Dict[str, Any]) -> Dict[str, Any]: """Remove sensitive keys from telemetry.""" sensitive_keys = ["password", "secret", "api_key"] return { k: v for k, v in data.items() if not any(sensitive in k.lower() for sensitive in sensitive_keys) } def normalize_timestamps(data: Dict[str, Any]) -> Dict[str, Any]: """Convert timestamp fields to ISO format.""" import time if "timestamp" in data and isinstance(data["timestamp"], (int, float)): from datetime import datetime data["timestamp"] = datetime.fromtimestamp(data["timestamp"]).isoformat() return data mcp = FastMCP(name="processor-demo") observability = instrument_server( mcp, config={ "server_name": "processor-demo", "server_version": "1.0.0", "exporter_auth": {"type": "bearer", "token": "token_xyz"}, "data_processors": [ add_environment_metadata, sanitize_sensitive_keys, normalize_timestamps ] } ) @mcp.tool() def process_request(data: dict) -> dict: """Process request with custom data processors applied to telemetry.""" return {"processed": True, "data": data} ``` -------------------------------- ### Conventional Commit Message Format (Example) Source: https://github.com/shinzo-labs/shinzo-py/blob/main/CONTRIBUTING.md Illustrates the structure of a conventional commit message, including type, scope, subject, optional body, and optional footer. This format is enforced by Commitizen and used for automatic versioning and changelog generation. ```markdown ``` (): [optional body] [optional footer] ``` **Types:** - **feat**: A new feature (triggers MINOR version bump, e.g., 1.0.0 → 1.1.0) - **fix**: A bug fix (triggers PATCH version bump, e.g., 1.0.0 → 1.0.1) - **docs**: Documentation only changes - **style**: Code style changes (formatting, missing semicolons, etc.) - **refactor**: Code refactoring without adding features or fixing bugs - **test**: Adding or updating tests - **chore**: Changes to build process or auxiliary tools **Scopes (optional):** Helps categorize commits (e.g., `instrumentation`, `config`, `exporters`) To trigger a MAJOR version bump (e.g., 1.0.0 → 2.0.0), add `!` after the type in your commit message. ```bash feat!: redesign configuration API BREAKING CHANGE: Configuration now uses a class-based approach instead of dictionaries. ``` ``` -------------------------------- ### Validate Telemetry Configuration with Python Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Validates telemetry configuration settings, ensuring required fields and authentication parameters are correctly specified. It uses Shinzo's ConfigValidator, TelemetryConfig, and AuthConfig. The example tests both valid and various invalid configurations, printing success messages or specific error details. ```python from shinzo import ConfigValidator, TelemetryConfig, AuthConfig try: # Valid configuration config = TelemetryConfig( server_name="validator-demo", server_version="1.0.0", sampling_rate=0.5, exporter_auth=AuthConfig( type="bearer", token="valid_token_123" ) ) ConfigValidator.validate(config) print("Configuration valid") except ValueError as e: print(f"Configuration error: {e}") # Test invalid configurations invalid_configs = [ # Missing server_name {"server_version": "1.0.0"}, # Invalid sampling rate {"server_name": "test", "server_version": "1.0.0", "sampling_rate": 1.5}, # Bearer auth without token { "server_name": "test", "server_version": "1.0.0", "exporter_auth": {"type": "bearer"} }, # API key auth without api_key { "server_name": "test", "server_version": "1.0.0", "exporter_auth": {"type": "apiKey"} }, # Basic auth without username/password { "server_name": "test", "server_version": "1.0.0", "exporter_auth": {"type": "basic", "username": "user"} } ] for invalid_config in invalid_configs: try: config = TelemetryConfig(**invalid_config) ConfigValidator.validate(config) except Exception as e: print(f"Expected error: {e}") ``` -------------------------------- ### Instrument Traditional MCP Server with Shinzo Source: https://github.com/shinzo-labs/shinzo-py/blob/main/README.md Illustrates how to instrument a traditional MCP server using the Shinzo SDK. This involves creating a Server instance and then using `instrument_server` with server configuration. It also includes defining an asynchronous tool and setting up a clean shutdown process. ```python from mcp.server import Server from shinzo import instrument_server # Create your MCP server server = Server("my-mcp-server") # Instrument it with Shinzo observability = instrument_server( server, config={ "server_name": "my-mcp-server", "server_version": "1.0.0", "exporter_auth": { "type": "bearer", "token": "your-api-token" } } ) # Define your tools @server.call_tool() async def get_weather(city: str) -> str: return f"Weather for {city}: Sunny" # Clean shutdown async def shutdown(): await observability.shutdown() ``` -------------------------------- ### Instrument FastMCP Server with Shinzo Source: https://github.com/shinzo-labs/shinzo-py/blob/main/README.md Demonstrates how to instrument a FastMCP server using the Shinzo SDK. It involves creating a FastMCP instance and then applying the `instrument_server` function with configuration details such as server name, version, and exporter authentication. ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server # Create FastMCP server mcp = FastMCP(name="my-mcp-server") # Instrument it with Shinzo observability = instrument_server( mcp, config={ "server_name": "my-mcp-server", "server_version": "1.0.0", "exporter_auth": { "type": "bearer", "token": "your-api-token" } } ) # Define your tools @mcp.tool() def get_weather(city: str) -> str: """Get weather for a city.""" return f"Weather for {city}: Sunny" # Run the server if __name__ == "__main__": mcp.run() ``` -------------------------------- ### Run Pytest Tests for Shinzo Py Source: https://github.com/shinzo-labs/shinzo-py/blob/main/README.md Execute the test suite for the shinzo-py library using pytest. Supports running all tests, tests with coverage reports, specific test files, and provides a make command for convenience. Validates configuration, PII sanitization, and authentication. ```bash # Run all tests python -m pytest tests/ # Run with coverage report python -m pytest tests/ --cov=shinzo --cov-report=term-missing # Run specific test file python -m pytest tests/test_config.py -v # Or use make make test ``` -------------------------------- ### Run Linting Checks for Shinzo Py Source: https://github.com/shinzo-labs/shinzo-py/blob/main/README.md Perform linting and code quality checks using make commands for the shinzo-py project. Includes formatting with black, linting with ruff, and type checking with mypy. Ensures code adheres to project standards before submission. ```bash # Run all linting checks (black, ruff, mypy) make lint-all # Or run individual checks make format # Format code with black make lint # Run ruff linter make type-check # Run mypy type checker ``` -------------------------------- ### Export Telemetry to Console for Local Development (Python) Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt This snippet demonstrates how to configure the Shinzo SDK to export telemetry data to the console. It's useful for debugging and local testing without requiring an external backend. The configuration includes enabling metrics, tracing, and argument collection, with a sampling rate of 1.0. ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server mcp = FastMCP(name="debug-server") observability = instrument_server( mcp, config={ "server_name": "debug-server", "server_version": "0.1.0", "exporter_type": "console", # Output to console "enable_metrics": True, "enable_tracing": True, "enable_argument_collection": True, "sampling_rate": 1.0 } ) @mcp.tool() def debug_operation(param1: str, param2: int) -> dict: """Operation for debugging telemetry output.""" # All spans and metrics will print to console return { "result": f"Processed {param1} with value {param2}", "timestamp": "2025-01-10T12:00:00Z" } @mcp.tool() def error_operation() -> dict: """Operation that raises an error for testing error tracking.""" raise ValueError("This is a test error for telemetry debugging") if __name__ == "__main__": # Run server - observe telemetry output in console mcp.run() ``` -------------------------------- ### Instrument Traditional MCP SDK Server Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Instruments traditional MCP SDK servers with OpenTelemetry tracing and metrics collection, providing similar observability features as with FastMCP. ```APIDOC ## POST /instrument_server (Traditional MCP SDK) ### Description Instruments traditional MCP SDK servers with OpenTelemetry tracing and metrics collection. This function enables comprehensive tracking of server activity, including tool calls, prompts, and performance metrics, with minimal configuration. ### Method POST ### Endpoint `/instrument_server` ### Parameters #### Query Parameters - **config** (object) - Required - Configuration object for telemetry collection and export settings. See `TelemetryConfig` for details. ### Request Body Similar to the FastMCP instrumentation, the `config` object is passed as an argument rather than a request body. ### Request Example ```python from mcp.server import Server from shinzo import instrument_server server = Server("data-processor") observability = instrument_server( server, config={ "server_name": "data-processor", "server_version": "2.1.0", "exporter_auth": { "type": "apiKey", "api_key": "api_key_xyz789" } } ) ``` ### Response #### Success Response (200) - **observability_instance** (object) - An instance of the observability handler, which can be used for shutdown or further customization. ``` -------------------------------- ### Instrument FastMCP Server with Shinzo Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Instruments a FastMCP server with Shinzo's observability features, including tracing, metrics, and configuration for telemetry export. It requires the `mcp.server.fastmcp` and `shinzo` libraries. The configuration specifies server details, exporter endpoint, authentication, and feature enablement. ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server # Create and instrument FastMCP server mcp = FastMCP(name="weather-service") observability = instrument_server( mcp, config={ "server_name": "weather-service", "server_version": "1.0.0", "exporter_endpoint": "https://api.app.shinzo.ai/telemetry/ingest_http", "exporter_auth": { "type": "bearer", "token": "sk_live_abc123xyz" }, "sampling_rate": 1.0, "enable_metrics": True, "enable_tracing": True, "enable_pii_sanitization": False, "enable_argument_collection": True, "metric_export_interval_ms": 60000, "batch_timeout_ms": 30000 } ) @mcp.tool() def get_forecast(city: str, days: int = 7) -> dict: """Get weather forecast for a city.""" return { "city": city, "forecast": [{"day": i, "temp": 20 + i, "condition": "sunny"} for i in range(days)] } if __name__ == "__main__": mcp.run() ``` -------------------------------- ### Manual Instrumentation for Custom Spans and Metrics in Python Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Enables advanced telemetry by creating custom spans, metrics (counters, histograms), and counters. This snippet uses FastMCP and shinzo, requiring definitions for `check_cache`. It demonstrates creating a counter for cache hits and a histogram for database query duration, along with custom spans for database operations. ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server mcp = FastMCP(name="custom-metrics-demo") observability = instrument_server( mcp, config={ "server_name": "custom-metrics-demo", "server_version": "1.0.0", "exporter_auth": {"type": "bearer", "token": "token_xyz"} } ) # Create custom metrics cache_hits = observability.get_increment_counter( "app.cache.hits", "Number of cache hits", "hits" ) db_query_duration = observability.get_histogram( "app.database.query.duration", "Database query execution time", "ms" ) @mcp.tool() async def fetch_user_data(user_id: str) -> dict: """Fetch user data with caching and custom metrics.""" cache_key = f"user:{user_id}" # Check cache cached_data = check_cache(cache_key) if cached_data: cache_hits(1, {"cache_type": "redis", "status": "hit"}) return cached_data # Create custom span for database operation async def db_operation(span): import time start = time.time() # Simulate database query user_data = {"user_id": user_id, "name": "John Doe", "email": "john@example.com"} duration_ms = (time.time() - start) * 1000 db_query_duration(duration_ms, {"query_type": "select", "table": "users"}) span.set_attribute("db.user_id", user_id) span.set_attribute("db.query.duration_ms", duration_ms) return user_data result = await observability.start_active_span( "fetch_user_from_database", {"db.system": "postgresql", "db.table": "users"}, db_operation ) return result def check_cache(key: str): # Simulate cache lookup return None ``` -------------------------------- ### Sanitize PII with Shinzo PIISanitizer Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Demonstrates using `PIISanitizer` from the Shinzo SDK to redact sensitive information from telemetry data. It allows defining custom regex patterns for identification and specifies a redaction string. Requires the `re` module and `shinzo.PIISanitizer`. ```python import re from shinzo import PIISanitizer # Create sanitizer with custom patterns custom_patterns = [ re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"), # Email re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), # SSN re.compile(r"\b\d{16}\b"), # Credit card re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), # IP address re.compile(r"\b[A-Z]{2}\d{6}\b") # Custom ID pattern ] sanitizer = PIISanitizer(patterns=custom_patterns, redact_value="[REDACTED]") # Sanitize telemetry data telemetry_data = { "user_email": "john.doe@example.com", "ip_address": "192.168.1.100", "ssn": "123-45-6789", "nested": { "credit_card": "1234567890123456", "custom_id": "AB123456" } } sanitized = sanitizer.sanitize(telemetry_data) # Output: { # "user_email": "[REDACTED]", # "ip_address": "[REDACTED]", # "ssn": "[REDACTED]", # "nested": { # "credit_card": "[REDACTED]", ``` -------------------------------- ### Instrument FastMCP Server Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Instruments a FastMCP server with OpenTelemetry tracing and metrics collection. This function automatically captures request/response data, execution duration, and error tracking. ```APIDOC ## POST /instrument_server (FastMCP) ### Description Instruments an MCP server built with the FastMCP framework with OpenTelemetry tracing and metrics collection. This function enables automatic capturing of telemetry data such as tool calls, prompts, resources, and performance metrics. ### Method POST ### Endpoint `/instrument_server` ### Parameters #### Query Parameters - **config** (object) - Required - Configuration object for telemetry collection and export settings. See `TelemetryConfig` for details. ### Request Body This endpoint does not typically have a request body in the traditional sense. The `config` object is passed as an argument. ### Request Example ```python from mcp.server.fastmcp import FastMCP from shinzo import instrument_server mcp = FastMCP(name="weather-service") observability = instrument_server( mcp, config={ "server_name": "weather-service", "server_version": "1.0.0", "exporter_endpoint": "https://api.app.shinzo.ai/telemetry/ingest_http", "exporter_auth": { "type": "bearer", "token": "sk_live_abc123xyz" }, "sampling_rate": 1.0, "enable_metrics": True, "enable_tracing": True, "enable_pii_sanitization": False, "enable_argument_collection": True, "metric_export_interval_ms": 60000, "batch_timeout_ms": 30000 } ) ``` ### Response #### Success Response (200) - **observability_instance** (object) - An instance of the observability handler for further configuration or shutdown. #### Response Example (The response is typically the observability instance, not a JSON payload in this context. The instrumentation is applied to the server object.) ``` -------------------------------- ### Define Shinzo Telemetry Configuration Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Defines a `TelemetryConfig` object for customizing Shinzo's observability settings. This includes server details, exporter configuration, authentication methods (basic auth shown here), PII sanitization, and data processors. It requires `shinzo.TelemetryConfig` and `shinzo.AuthConfig`. ```python from shinzo import TelemetryConfig, AuthConfig config = TelemetryConfig( server_name="analytics-api", server_version="3.0.1", exporter_endpoint="https://otel-collector.company.com/v1/traces", exporter_auth=AuthConfig( type="basic", username="telemetry_user", password="secure_password_123" ), sampling_rate=0.25, enable_pii_sanitization=True, enable_argument_collection=True, metric_export_interval_ms=30000, batch_timeout_ms=15000, exporter_type="otlp-http", enable_metrics=True, enable_tracing=True, data_processors=[ lambda data: {**data, "environment": "production"} ] ) # Use with server instrumentation from mcp.server.fastmcp import FastMCP from shinzo import instrument_server mcp = FastMCP(name="analytics-api") observability = instrument_server(mcp, config=config) ``` -------------------------------- ### TelemetryConfig Model Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Defines the configuration model for telemetry collection and export settings, including server details, exporter configuration, sampling rates, and feature flags. ```APIDOC ## TelemetryConfig ### Description The `TelemetryConfig` model encapsulates all settings required for configuring the Shinzo observability SDK. This includes details about the server, how and where to export telemetry data, and which features to enable or disable. ### Fields - **server_name** (str) - Required - The name of the server emitting telemetry. - **server_version** (str) - Optional - The version of the server. - **exporter_endpoint** (str) - Optional - The URL of the OpenTelemetry collector or backend endpoint. - **exporter_auth** (AuthConfig) - Optional - Authentication configuration for the exporter. - **sampling_rate** (float) - Optional - The rate at which telemetry data should be sampled (0.0 to 1.0). Defaults to 1.0 (no sampling). - **enable_metrics** (bool) - Optional - Whether to enable metrics collection. Defaults to True. - **enable_tracing** (bool) - Optional - Whether to enable distributed tracing. Defaults to True. - **enable_pii_sanitization** (bool) - Optional - Whether to enable PII sanitization. Defaults to False. - **enable_argument_collection** (bool) - Optional - Whether to collect function arguments in traces. Defaults to True. - **metric_export_interval_ms** (int) - Optional - The interval in milliseconds for exporting metrics. Defaults to 60000. - **batch_timeout_ms** (int) - Optional - The timeout in milliseconds for batching telemetry data before exporting. Defaults to 30000. - **exporter_type** (str) - Optional - The type of exporter to use (e.g., 'otlp-http'). Defaults to 'otlp-http'. - **data_processors** (list) - Optional - A list of functions to process telemetry data before exporting. ### Example Usage ```python from shinzo import TelemetryConfig, AuthConfig config = TelemetryConfig( server_name="analytics-api", server_version="3.0.1", exporter_endpoint="https://otel-collector.company.com/v1/traces", exporter_auth=AuthConfig( type="basic", username="telemetry_user", password="secure_password_123" ), sampling_rate=0.25, enable_pii_sanitization=True, enable_argument_collection=True, metric_export_interval_ms=30000, batch_timeout_ms=15000, exporter_type="otlp-http", enable_metrics=True, enable_tracing=True, data_processors=[ lambda data: {**data, "environment": "production"} ] ) ``` ``` -------------------------------- ### PIISanitizer Class Source: https://context7.com/shinzo-labs/shinzo-py/llms.txt Provides functionality to sanitize personally identifiable information (PII) from telemetry data using configurable regex patterns. ```APIDOC ## PIISanitizer ### Description The `PIISanitizer` class allows for the detection and redaction of sensitive information within telemetry data. It supports custom regex patterns for flexible PII identification and allows specifying a redaction string. ### Methods #### `__init__(self, patterns: list[re.Pattern], redact_value: str = "[REDACTED]")` Initializes the `PIISanitizer` with a list of compiled regular expression patterns and an optional redaction value. - **patterns** (list[re.Pattern]) - Required - A list of compiled regex patterns to identify PII. - **redact_value** (str) - Optional - The string to replace detected PII with. Defaults to "[REDACTED]". #### `sanitize(self, data: dict) -> dict` Sanitizes the input dictionary by applying the defined PII patterns. - **data** (dict) - Required - The dictionary containing telemetry data to be sanitized. - **Returns** (dict) - A new dictionary with PII redacted. ### Request Example ```python import re from shinzo import PIISanitizer # Create sanitizer with custom patterns custom_patterns = [ re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"), # Email re.compile(r"\b\d{3}-\d{2}-\d{4}\b"), # SSN re.compile(r"\b\d{16}\b"), # Credit card re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), # IP address re.compile(r"\b[A-Z]{2}\d{6}\b") # Custom ID pattern ] sanitizer = PIISanitizer(patterns=custom_patterns, redact_value="[REDACTED]") # Sanitize telemetry data telemetry_data = { "user_email": "john.doe@example.com", "ip_address": "192.168.1.100", "ssn": "123-45-6789", "nested": { "credit_card": "1234567890123456", "custom_id": "AB123456" } } sanitized = sanitizer.sanitize(telemetry_data) ``` ### Response #### Success Response (200) - **sanitized_data** (dict) - The telemetry data with detected PII replaced by the redaction value. #### Response Example ```json { "user_email": "[REDACTED]", "ip_address": "[REDACTED]", "ssn": "[REDACTED]", "nested": { "credit_card": "[REDACTED]", "custom_id": "[REDACTED]" } } ``` ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.