add observability to mcp agent through observability manager

Author: Pederzh

src/agents/mcp_agent.ts

@@ -1,3 +1,4 @@
+import type { BaseCallbackHandler } from '@langchain/core/callbacks/base'
 import type { BaseLanguageModelInterface, LanguageModelLike } from '@langchain/core/language_models/base'
 import type {
   BaseMessage,
@@ -29,6 +30,7 @@ import { zodToJsonSchema } from 'zod-to-json-schema'
 import { LangChainAdapter } from '../adapters/langchain_adapter.js'
 import { logger } from '../logging.js'
 import { ServerManager } from '../managers/server_manager.js'
+import { ObservabilityManager } from '../observability/index.js'
 import { extractModelInfo, Telemetry } from '../telemetry/index.js'
 import { createSystemMessage } from './prompts/system_prompt_builder.js'
 import { DEFAULT_SYSTEM_PROMPT_TEMPLATE, SERVER_MANAGER_SYSTEM_PROMPT_TEMPLATE } from './prompts/templates.js'
@@ -61,6 +63,10 @@ export class MCPAgent {
   private modelProvider: string
   private modelName: string
 
+  // Observability support
+  private observabilityManager: ObservabilityManager
+  private callbacks: BaseCallbackHandler[] = []
+
   // Remote agent support
   private isRemote = false
   private remoteAgent: RemoteAgent | null = null
@@ -81,6 +87,7 @@ export class MCPAgent {
     verbose?: boolean
     adapter?: LangChainAdapter
     serverManagerFactory?: (client: MCPClient) => ServerManager
+    callbacks?: BaseCallbackHandler[]
     // Remote agent parameters
     agentId?: string
     apiKey?: string
@@ -107,6 +114,8 @@ export class MCPAgent {
       this.telemetry = Telemetry.getInstance()
       this.modelProvider = 'remote'
       this.modelName = 'remote-agent'
+      this.observabilityManager = new ObservabilityManager({ customCallbacks: options.callbacks })
+      this.callbacks = []
       return
     }
 
@@ -159,6 +168,12 @@ export class MCPAgent {
       this.modelName = 'unknown'
     }
 
+    // Set up observability callbacks using the ObservabilityManager
+    this.observabilityManager = new ObservabilityManager({
+      customCallbacks: options.callbacks,
+      verbose: this.verbose,
+    })
+
     // Make getters configurable for test mocking
     Object.defineProperty(this, 'agentExecutor', {
       get: () => this._agentExecutor,
@@ -183,6 +198,13 @@ export class MCPAgent {
 
     logger.info('🚀 Initializing MCP agent and connecting to services...')
 
+    // Initialize observability callbacks
+    this.callbacks = await this.observabilityManager.getCallbacks()
+    const handlerNames = await this.observabilityManager.getHandlerNames()
+    if (handlerNames.length > 0) {
+      logger.info(`📊 Observability enabled with: ${handlerNames.join(', ')}`)
+    }
+
     // If using server manager, initialize it
     if (this.useServerManager && this.serverManager) {
       await this.serverManager.initialize()
@@ -202,7 +224,7 @@ export class MCPAgent {
       // Standard initialization - if using client, get or create sessions
       if (this.client) {
         // First try to get existing sessions
-        this.sessions = await this.client.getAllActiveSessions()
+        this.sessions = this.client.getAllActiveSessions()
         logger.info(`🔌 Found ${Object.keys(this.sessions).length} existing sessions`)
 
         // If no active sessions exist, create new ones
@@ -294,6 +316,7 @@ export class MCPAgent {
       maxIterations: this.maxSteps,
       verbose: this.verbose,
       returnIntermediateSteps: true,
+      callbacks: this.callbacks,
     })
   }
 
@@ -642,7 +665,7 @@ export class MCPAgent {
 
       let serverCount = 0
       if (this.client) {
-        serverCount = Object.keys(await this.client.getAllActiveSessions()).length
+        serverCount = Object.keys(this.client.getAllActiveSessions()).length
       }
       else if (this.connectors) {
         serverCount = this.connectors.length
@@ -690,6 +713,9 @@ export class MCPAgent {
     }
 
     logger.info('🔌 Closing MCPAgent resources…')
+
+    // Shutdown observability handlers (important for serverless)
+    await this.observabilityManager.shutdown()
     try {
       this._agentExecutor = null
       this._tools = []
@@ -779,7 +805,10 @@ export class MCPAgent {
       // Stream events from the agent executor
       const eventStream = agentExecutor.streamEvents(
         inputs,
-        { version: 'v2' },
+        {
+          version: 'v2',
+          callbacks: this.callbacks.length > 0 ? this.callbacks : undefined,
+        },
       )
 
       // Yield each event
@@ -829,7 +858,7 @@ export class MCPAgent {
 
       let serverCount = 0
       if (this.client) {
-        serverCount = Object.keys(await this.client.getAllActiveSessions()).length
+        serverCount = Object.keys(this.client.getAllActiveSessions()).length
       }
       else if (this.connectors) {
         serverCount = this.connectors.length

src/observability/README.md

@@ -0,0 +1,199 @@
+# Observability Module
+
+This module provides comprehensive observability for MCP agents using LangChain, supporting multiple observability platforms including Langfuse and Laminar.
+
+## Features
+
+- **Multi-platform support**: Integrates with Langfuse and Laminar
+- **Automatic instrumentation**: Zero-code observability with environment variables
+- **Custom callbacks**: Support for custom LangChain callback handlers
+- **TypeScript-first**: Full TypeScript support with proper types
+- **Serverless-ready**: Proper shutdown handling for serverless environments
+
+## Installation
+
+Install the required packages based on your observability platform:
+
+```bash
+# For Langfuse
+npm install langfuse langfuse-langchain
+
+# For Laminar
+npm install @lmnr-ai/lmnr
+
+# Or install both
+npm install langfuse langfuse-langchain @lmnr-ai/lmnr
+```
+
+## Configuration
+
+### Langfuse
+
+Set the following environment variables:
+
+```env
+LANGFUSE_PUBLIC_KEY=pk-lf-...
+LANGFUSE_SECRET_KEY=sk-lf-...
+LANGFUSE_HOST=https://cloud.langfuse.com  # Optional
+MCP_USE_LANGFUSE=true  # Set to false to disable
+```
+
+### Laminar
+
+Set the following environment variables:
+
+```env
+LAMINAR_PROJECT_API_KEY=your_project_api_key
+# or
+LMNR_PROJECT_API_KEY=your_project_api_key
+MCP_USE_LAMINAR=true  # Set to false to disable
+```
+
+## Usage
+
+### Basic Usage with MCPAgent
+
+The observability is automatically integrated into MCPAgent:
+
+```typescript
+import { MCPAgent } from 'mcp-use'
+
+const agent = new MCPAgent({
+  llm: myLLM,
+  client: myMCPClient,
+  // Observability is automatically enabled if environment variables are set
+})
+
+// Initialize the agent
+await agent.initialize()
+
+// Run queries - they will be automatically traced
+const result = await agent.run('What\'s the weather?')
+```
+
+### Custom Callbacks
+
+You can provide custom callbacks:
+
+```typescript
+import { CallbackHandler } from 'langfuse-langchain'
+import { MCPAgent } from 'mcp-use'
+
+const customHandler = new CallbackHandler({
+  secretKey: 'custom-secret',
+  publicKey: 'custom-public',
+})
+
+const agent = new MCPAgent({
+  llm: myLLM,
+  client: myMCPClient,
+  callbacks: [customHandler], // Use custom callbacks instead of auto-detected ones
+})
+```
+
+### Direct ObservabilityManager Usage
+
+For advanced use cases, you can use the ObservabilityManager directly:
+
+```typescript
+import { ObservabilityManager } from 'mcp-use/observability'
+
+// Create a manager
+const manager = new ObservabilityManager({
+  verbose: true, // Enable verbose logging
+})
+
+// Get available callbacks
+const callbacks = await manager.getCallbacks()
+
+// Check available handlers
+const handlerNames = await manager.getHandlerNames()
+console.log('Available handlers:', handlerNames)
+
+// Add custom callback
+manager.addCallback(myCustomCallback)
+
+// Shutdown (important for serverless)
+await manager.shutdown()
+```
+
+## Platform Features
+
+### Langfuse Features
+
+- Detailed LLM call tracing
+- Chain execution tracking
+- Tool usage monitoring
+- Cost tracking
+- Custom metadata and tags
+- Session management
+- User tracking
+
+### Laminar Features
+
+- Automatic OpenTelemetry instrumentation
+- Zero-code tracing
+- Real-time monitoring
+- Custom spans with `@observe`
+- Evaluation framework
+- Session grouping
+
+## Serverless Considerations
+
+For serverless environments (AWS Lambda, Vercel, etc.), ensure proper shutdown:
+
+```typescript
+const agent = new MCPAgent({ /* ... */ })
+
+try {
+  await agent.initialize()
+  const result = await agent.run(query)
+  return result
+}
+finally {
+  // Important: Ensure traces are flushed
+  await agent.close()
+}
+```
+
+## Debugging
+
+Enable debug logging to see observability events:
+
+```typescript
+import { logger } from 'mcp-use/logging'
+
+// Set log level to debug
+process.env.LOG_LEVEL = 'debug'
+
+// Now you'll see detailed observability logs
+```
+
+## Environment Variables Reference
+
+### Langfuse
+
+- `LANGFUSE_PUBLIC_KEY` - Required: Your Langfuse public key
+- `LANGFUSE_SECRET_KEY` - Required: Your Langfuse secret key
+- `LANGFUSE_HOST` / `LANGFUSE_BASEURL` - Optional: Langfuse API URL (default: https://cloud.langfuse.com)
+- `LANGFUSE_RELEASE` - Optional: Release/version identifier
+- `LANGFUSE_FLUSH_AT` - Optional: Batch size for flushing (default: 15)
+- `LANGFUSE_FLUSH_INTERVAL` - Optional: Flush interval in ms (default: 10000)
+- `LANGFUSE_REQUEST_TIMEOUT` - Optional: Request timeout in ms (default: 10000)
+- `LANGFUSE_ENABLED` - Optional: Set to "false" to disable
+- `MCP_USE_LANGFUSE` - Optional: Set to "false" to disable Langfuse integration
+
+### Laminar
+
+- `LAMINAR_PROJECT_API_KEY` / `LMNR_PROJECT_API_KEY` - Required: Your Laminar project API key
+- `LAMINAR_BASE_URL` / `LMNR_BASE_URL` - Optional: Laminar API URL
+- `MCP_USE_LAMINAR` - Optional: Set to "false" to disable Laminar integration
+
+## Examples
+
+See the [examples](../../examples/) directory for complete working examples:
+
+- Basic observability setup
+- Multi-platform configuration
+- Custom callback handlers
+- Serverless deployments

src/observability/index.ts

@@ -0,0 +1,30 @@
+/**
+ * Observability module for MCP-use.
+ *
+ * This module provides centralized observability management for LangChain agents,
+ * supporting multiple platforms like Langfuse and Laminar.
+ */
+
+// Import observability providers - order matters for initialization
+import './laminar.js'
+import './langfuse.js'
+
+// Re-export individual handlers for direct usage if needed
+export {
+  getLaminarHandler,
+  getLaminarInitPromise,
+  isLaminarInitialized,
+} from './laminar.js'
+export {
+  langfuseClient,
+  langfuseHandler,
+  langfuseInitPromise,
+} from './langfuse.js'
+
+// Export the manager and its utilities
+export {
+  createManager,
+  getDefaultManager,
+  type ObservabilityConfig,
+  ObservabilityManager,
+} from './manager.js'