Unstructured API MCP Server

Instructions

List available sources from the Unstructured API.

Args:
    source_type: Optional source connector type to filter by

Returns:
    String containing the list of sources

Implementation Reference

• uns_mcp/server.py:113-152 (handler)

  The main handler function for the 'list_sources' MCP tool, decorated with @mcp.tool(). It retrieves the list of source connectors from the Unstructured API client, optionally filters by source_type, sorts them, and returns a formatted string listing the sources with their names and IDs.

  @mcp.tool()
  async def list_sources(
      ctx: Context,
      source_type: Optional[SourceConnectorType | str] = None,
  ) -> str:
      """
      List available sources from the Unstructured API.
  
      Args:
          source_type: Optional source connector type to filter by
  
      Returns:
          String containing the list of sources
      """
      client = ctx.request_context.lifespan_context.client
  
      request = ListSourcesRequest()
      if source_type:
          try:
              source_type = (
                  SourceConnectorType(source_type) if isinstance(source_type, str) else source_type
              )
              request.source_type = source_type
          except KeyError:
              return f"Invalid source type: {source_type}"
  
      response = await client.sources.list_sources_async(request=request)
  
      # Sort sources by name
      sorted_sources = sorted(response.response_list_sources, key=lambda source: source.name.lower())
  
      if not sorted_sources:
          return "No sources found"
  
      # Format response
      result = ["Available sources:"]
      for source in sorted_sources:
          result.append(f"- {source.name} (ID: {source.id})")
  
      return "\n".join(result)