Emit structured product events for MCP sessions and tool calls.
| 20 | |
| 21 | |
| 22 | class EventLoggingMiddleware(Middleware): |
| 23 | """Emit structured product events for MCP sessions and tool calls.""" |
| 24 | |
| 25 | async def on_initialize( |
| 26 | self, |
| 27 | context: MiddlewareContext[mt.InitializeRequest], |
| 28 | call_next: CallNext[mt.InitializeRequest, mt.InitializeResult | None], |
| 29 | ) -> mt.InitializeResult | None: |
| 30 | result = await call_next(context) |
| 31 | client_info = context.message.params.clientInfo |
| 32 | logger.info( |
| 33 | "session.opened", |
| 34 | flagsmith__mcp__client__name=client_info.name, |
| 35 | flagsmith__mcp__client__version=client_info.version, |
| 36 | ) |
| 37 | return result |
| 38 | |
| 39 | async def on_call_tool( |
| 40 | self, |
| 41 | context: MiddlewareContext[mt.CallToolRequestParams], |
| 42 | call_next: CallNext[mt.CallToolRequestParams, ToolResult], |
| 43 | ) -> ToolResult: |
| 44 | client_info = get_client_info() |
| 45 | client_name = client_info.name if client_info else "" |
| 46 | client_version = client_info.version if client_info else "" |
| 47 | try: |
| 48 | result = await call_next(context) |
| 49 | except Exception: |
| 50 | logger.info( |
| 51 | "tool.called", |
| 52 | tool__name=context.message.name, |
| 53 | flagsmith__mcp__client__name=client_name, |
| 54 | flagsmith__mcp__client__version=client_version, |
| 55 | status="error", |
| 56 | ) |
| 57 | raise |
| 58 | logger.info( |
| 59 | "tool.called", |
| 60 | tool__name=context.message.name, |
| 61 | flagsmith__mcp__client__name=client_name, |
| 62 | flagsmith__mcp__client__version=client_version, |
| 63 | status="success", |
| 64 | ) |
| 65 | return result |
no outgoing calls
no test coverage detected
searching dependent graphs…