list_flows
Extract detailed HTTP request/response data, headers, content, and metadata from specific session flows using the mitmproxy-mcp MCP Server.
Instructions
Retrieves detailed HTTP request/response data including headers, content (or structure preview for large JSON), and metadata from specified flows
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | The ID of the session to list flows from |
Implementation Reference
- src/mitmproxy_mcp/server.py:133-162 (handler)The primary handler function for the 'list_flows' tool. It extracts the session_id from arguments, loads flows using the helper get_flows_from_dump, filters HTTP flows, and returns a JSON list of flow summaries including index, method, URL, and status code.async def list_flows(arguments: dict) -> list[types.TextContent]: """ Lists HTTP flows from a mitmproxy dump file. """ session_id = arguments.get("session_id") if not session_id: return [types.TextContent(type="text", text="Error: Missing session_id")] try: flows = await get_flows_from_dump(session_id) flow_list = [] for i, flow in enumerate(flows): if flow.type == "http": request = flow.request response = flow.response flow_info = { "index": i, "method": request.method, "url": request.url, "status": response.status_code if response else None } flow_list.append(flow_info) return [types.TextContent(type="text", text=json.dumps(flow_list, indent=2))] except FileNotFoundError: return [types.TextContent(type="text", text="Error: Session not found")] except Exception as e: return [types.TextContent(type="text", text=f"Error reading flows: {str(e)}")]
- src/mitmproxy_mcp/server.py:41-50 (schema)JSON Schema defining the input parameters for the 'list_flows' tool, requiring a 'session_id' string.inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "The ID of the session to list flows from" } }, "required": ["session_id"] }
- src/mitmproxy_mcp/server.py:38-51 (registration)Tool object creation and registration within the @server.list_tools() handler, specifying name, description, and schema.types.Tool( name="list_flows", description="Retrieves detailed HTTP request/response data including headers, content (or structure preview for large JSON), and metadata from specified flows", inputSchema={ "type": "object", "properties": { "session_id": { "type": "string", "description": "The ID of the session to list flows from" } }, "required": ["session_id"] } ),
- src/mitmproxy_mcp/server.py:410-411 (registration)Dispatch logic in @server.call_tool() handler that routes 'list_flows' calls to the handler function.if name == "list_flows": return await list_flows(arguments)
- Helper function called by the list_flows handler to asynchronously load and cache mitmproxy flows from the session dump file.async def get_flows_from_dump(session_id: str) -> list: """ Retrieves flows from the dump file, using the cache if available. """ dump_file = os.path.join(DUMP_DIR, f"{session_id}.dump") if not os.path.exists(dump_file): raise FileNotFoundError("Session not found") if session_id in FLOW_CACHE: return FLOW_CACHE[session_id] else: with open(dump_file, "rb") as f: reader = io.FlowReader(f) flows = list(reader.stream()) FLOW_CACHE[session_id] = flows return flows