list_flows
Retrieve detailed HTTP request/response data including headers, content previews, and metadata from a specific mitmproxy session to analyze network traffic.
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 processes the session_id argument, loads flows from the dump file using a helper function, iterates over HTTP flows to create summaries (index, method, URL, status), and returns a JSON-formatted list or appropriate error messages.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:38-51 (registration)Registration of the 'list_flows' tool in the @server.list_tools() handler, defining the tool name, description, and input JSON schema requiring 'session_id'.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 the @server.call_tool() handler that routes calls to the 'list_flows' function when the tool name matches.if name == "list_flows": return await list_flows(arguments)
- Supporting helper function called by the list_flows handler to load and cache mitmproxy flows from the session dump file using mitmproxy's FlowReader.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