list_sources
Retrieve available source connectors from the Unstructured API, optionally filtering by a specific type to streamline data integration and workflow management.
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
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source_type | No |
Implementation Reference
- uns_mcp/server.py:113-153 (handler)The primary handler function for the 'list_sources' MCP tool. It retrieves the list of available source connectors from the Unstructured API client, optionally filters by source_type, sorts them by name, and returns a formatted string list.@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)