RequestRepo MCP

Instructions

Implementation Reference

• src/requestrepo_mcp/server.py:168-185 (handler)

  The core business logic handler that retrieves a shared request by token, serializes it with options for including raw data and body content, and returns the structured response.

  def get_shared_request(
      self,
      *,
      share_token: str,
      include_raw: bool = False,
      include_body: bool = False,
      max_bytes: int | None = None,
  ) -> dict[str, Any]:
      request = self._client().get_shared_request(share_token)
      return {
          "share_token": share_token,
          "request": serialize_request(
              request,
              include_raw=include_raw,
              include_body=include_body,
              max_bytes=self._resolved_max_bytes(max_bytes),
          ),
      }

• src/requestrepo_mcp/server.py:397-410 (registration)

  MCP tool registration that exposes the get_shared_request functionality with the @mcp.tool() decorator, defining the tool's parameters and delegation to the service layer.

  @mcp.tool()
  def get_shared_request(
      share_token: str,
      include_raw: bool = False,
      include_body: bool = False,
      max_bytes: int = 65536,
  ) -> dict[str, Any]:
      """Resolve a shared request token."""
      return resolved_service.get_shared_request(
          share_token=share_token,
          include_raw=include_raw,
          include_body=include_body,
          max_bytes=max_bytes,
      )

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

  Serialization helper that converts request objects (HttpRequest, DnsRequest, SmtpRequest, TcpRequest) into dictionary format with optional raw data and body content, handling byte truncation based on max_bytes limit.

  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