Prefect MCP Server

Instructions

Implementation Reference

• src/mcp_prefect/block.py:66-103 (handler)

  The main handler function for the 'get_block_documents' tool. It is decorated with @mcp.tool, which registers it as an MCP tool. The function retrieves block documents from the Prefect client using filters based on block_type_slug, optional limit, offset, and name pattern, formats the result as text content, and returns it.

  @mcp.tool
  async def get_block_documents(
      block_type_slug: str,
      limit: Optional[int] = None,
      offset: Optional[int] = None,
      name: Optional[str] = None,
  ) -> List[Union[types.TextContent, types.ImageContent, types.EmbeddedResource]]:
      """
      Get block documents by block type.
      
      Args:
          block_type_slug: The block type slug
          limit: Maximum number of block documents to return
          offset: Number of block documents to skip
          name: Filter by name pattern
          
      Returns:
          A list of block documents with their details
      """
      async with get_client() as client:
          # Build filter parameters
          filters = {}
          if block_type_slug:
              filters["block_type_slug"] = {"eq_": block_type_slug}
          if name:
              filters["name"] = {"like_": f"%{name}%"}
          
          block_documents = await client.read_block_documents(
              limit=limit,
              offset=offset,
              **filters
          )
          
          block_documents_result = {
              "block_documents": [block_doc.model_dump() for block_doc in block_documents]
          }
          
          return [types.TextContent(type="text", text=str(block_documents_result))]

• src/mcp_prefect/block.py:66-66 (registration)

  The @mcp.tool decorator registers the get_block_documents function as an MCP tool.

  @mcp.tool