RequestRepo MCP

Instructions

Implementation Reference

• src/requestrepo_mcp/server.py:58-92 (handler)

  The core business logic handler for list_requests tool. Validates parameters (limit, offset, max_bytes), resolves configuration, calls the requestrepo client's list_requests method, optionally filters by request_type, and serializes the results using serialize_request.

  def list_requests(
      self,
      *,
      limit: int = 100,
      offset: int = 0,
      request_type: RequestType | None = None,
      include_raw: bool = False,
      include_body: bool = False,
      max_bytes: int | None = None,
  ) -> dict[str, Any]:
      if limit < 1:
          raise ValueError("limit must be >= 1.")
      if offset < 0:
          raise ValueError("offset must be >= 0.")
  
      resolved_max_bytes = self._resolved_max_bytes(max_bytes)
      requests = self._client().list_requests(limit=limit, offset=offset)
      if request_type is not None:
          requests = [request for request in requests if request.type == request_type]
  
      return {
          "limit": limit,
          "offset": offset,
          "request_type": request_type,
          "count": len(requests),
          "requests": [
              serialize_request(
                  request,
                  include_raw=include_raw,
                  include_body=include_body,
                  max_bytes=resolved_max_bytes,
              )
              for request in requests
          ],
      }

• src/requestrepo_mcp/server.py:344-361 (registration)

  MCP tool registration using @mcp.tool() decorator. This exposes the list_requests functionality as an MCP tool with parameters including limit, offset, request_type, include_raw, include_body, and max_bytes.

  @mcp.tool()
  def list_requests(
      limit: int = 100,
      offset: int = 0,
      request_type: RequestType | None = None,
      include_raw: bool = False,
      include_body: bool = False,
      max_bytes: int = 65536,
  ) -> dict[str, Any]:
      """List captured requests with optional filtering."""
      return resolved_service.list_requests(
          limit=limit,
          offset=offset,
          request_type=request_type,
          include_raw=include_raw,
          include_body=include_body,
          max_bytes=max_bytes,
      )

• src/requestrepo_mcp/schemas.py:9-9 (schema)

  RequestType schema definition used for filtering requests by type (http, dns, smtp, tcp). This type is used as the request_type parameter in the list_requests tool.

  RequestType = Literal["http", "dns", "smtp", "tcp"]

• src/requestrepo_mcp/serializers.py:74-129 (helper)

  Helper function that serializes request objects (HttpRequest, DnsRequest, SmtpRequest, TcpRequest) into dictionaries. Called by list_requests handler to format each request with optional raw data and body content based on include_raw and include_body flags.

  def serialize_request(
      request: HttpRequest | DnsRequest | SmtpRequest | TcpRequest,
      *,
      include_raw: bool,
      include_body: bool,
      max_bytes: int,
  ) -> dict[str, Any]:
      payload: dict[str, Any] = {
          "id": request.id,
          "type": request.type,
          "uid": request.uid,
          "ip": request.ip,
          "country": request.country,
          "date_unix": request.date,
          "date_iso": _iso_from_unix(request.date),
      }
  
      if isinstance(request, HttpRequest):
          payload.update(
              {
                  "method": request.method,
                  "path": request.path,
                  "http_version": request.http_version,
                  "headers": request.headers,
              }
          )
          if include_body and request.body is not None:
              payload["body"] = bytes_envelope(request.body, max_bytes=max_bytes)
      elif isinstance(request, DnsRequest):
          payload.update(
              {
                  "port": request.port,
                  "query_type": request.query_type,
                  "domain": request.domain,
                  "reply": request.reply,
              }
          )
      elif isinstance(request, SmtpRequest):
          payload.update(
              {
                  "command": request.command,
                  "data": request.data,
                  "subject": request.subject,
                  "from_addr": request.from_addr,
                  "to": request.to,
                  "cc": request.cc,
                  "bcc": request.bcc,
              }
          )
      elif isinstance(request, TcpRequest):
          payload.update({"port": request.port})
  
      if include_raw:
          payload["raw"] = bytes_envelope(request.raw, max_bytes=max_bytes)
  
      return payload