get_flow

Instructions

Implementation Reference

• tools/flow_tools.py:236-277 (handler)

  The MCP tool handler for 'get_flow'. Defines the async function decorated with @mcp.tool() that accepts flow_id (required) and doc_type (optional, defaults to 'Metadata'). Calls the HTTP client, and if the result is bytes (binary doc types), encodes it as base64 in the response dict.

  @mcp.tool()
  async def get_flow(
      flow_id: Annotated[
          str,
          Field(
              description=(
                  "Flow identifier assigned by the Approved Platform "
                  "(returned by submit_flow or search_flows, maxLength 36)."
              )
          ),
      ],
      doc_type: Annotated[
          str,
          Field(
              default="Metadata",
              description=(
                  "Document type to retrieve: "
                  "Metadata (default, returns the flow's JSON metadata — recommended), "
                  "Original (original submitted document, returned as base64), "
                  "Converted (document converted by the AP, returned as base64), "
                  "ReadableView (human-readable PDF representation, returned as base64)."
              ),
          ),
      ] = "Metadata",
  ) -> dict:
      """
      Retrieve a flow by its identifier. docType allows choosing between
      JSON metadata (Metadata), the original document (Original), the converted
      document (Converted), or the readable representation (ReadableView).
      By default, returns the JSON metadata (status, dates, identifiers).
      """
      client = get_flow_client()
      result = await client.get_flow(flow_id=flow_id, doc_type=doc_type)
  
      if isinstance(result, bytes):
          # Encode as base64 for JSON serialisation
          return {
              "flowId": flow_id,
              "docType": doc_type,
              "contentBase64": base64.b64encode(result).decode(),
          }
      return result

• tools/flow_tools.py:236-259 (schema)

  Pydantic field definitions for the get_flow tool. flow_id is required (string, maxLength 36). doc_type is optional with a default of 'Metadata', allowing values Metadata, Original, Converted, or ReadableView.

  @mcp.tool()
  async def get_flow(
      flow_id: Annotated[
          str,
          Field(
              description=(
                  "Flow identifier assigned by the Approved Platform "
                  "(returned by submit_flow or search_flows, maxLength 36)."
              )
          ),
      ],
      doc_type: Annotated[
          str,
          Field(
              default="Metadata",
              description=(
                  "Document type to retrieve: "
                  "Metadata (default, returns the flow's JSON metadata — recommended), "
                  "Original (original submitted document, returned as base64), "
                  "Converted (document converted by the AP, returned as base64), "
                  "ReadableView (human-readable PDF representation, returned as base64)."
              ),
          ),
      ] = "Metadata",

• tools/flow_tools.py:36-41 (helper)

  Helper function that returns a shared singleton FlowClient instance, used by get_flow and all other flow tools.

  def get_flow_client() -> FlowClient:
      global _flow_client
      if _flow_client is None:
          _flow_client = FlowClient()
      return _flow_client

• clients/flow_client.py:156-165 (helper)

  The HTTP client method `FlowClient.get_flow()` that makes the actual GET /v1/flows/{flowId} request. For doc_type='Metadata' it returns parsed JSON, otherwise returns raw bytes content.

  async def get_flow(
      self, flow_id: str, doc_type: str = "Metadata"
  ) -> dict[str, Any] | bytes:
      """GET /v1/flows/{flowId} — Retrieve a flow by identifier."""
      response = await self._request(
          "GET", f"/v1/flows/{flow_id}", params={"docType": doc_type}
      )
      if doc_type == "Metadata":
          return response.json()
      return response.content

• server.py:63-64 (registration)

  The top-level registration call that passes the FastMCP instance to register_flow_tools, which decorates the get_flow function with @mcp.tool().

  register_flow_tools(mcp)
  register_directory_tools(mcp)