This is page 20 of 35. Use http://codebase.md/dicklesworthstone/llm_gateway_mcp_server?lines=false&page={x} to view the full context.
# Directory Structure
```
├── .cursorignore
├── .env.example
├── .envrc
├── .gitignore
├── additional_features.md
├── check_api_keys.py
├── completion_support.py
├── comprehensive_test.py
├── docker-compose.yml
├── Dockerfile
├── empirically_measured_model_speeds.json
├── error_handling.py
├── example_structured_tool.py
├── examples
│ ├── __init__.py
│ ├── advanced_agent_flows_using_unified_memory_system_demo.py
│ ├── advanced_extraction_demo.py
│ ├── advanced_unified_memory_system_demo.py
│ ├── advanced_vector_search_demo.py
│ ├── analytics_reporting_demo.py
│ ├── audio_transcription_demo.py
│ ├── basic_completion_demo.py
│ ├── cache_demo.py
│ ├── claude_integration_demo.py
│ ├── compare_synthesize_demo.py
│ ├── cost_optimization.py
│ ├── data
│ │ ├── sample_event.txt
│ │ ├── Steve_Jobs_Introducing_The_iPhone_compressed.md
│ │ └── Steve_Jobs_Introducing_The_iPhone_compressed.mp3
│ ├── docstring_refiner_demo.py
│ ├── document_conversion_and_processing_demo.py
│ ├── entity_relation_graph_demo.py
│ ├── filesystem_operations_demo.py
│ ├── grok_integration_demo.py
│ ├── local_text_tools_demo.py
│ ├── marqo_fused_search_demo.py
│ ├── measure_model_speeds.py
│ ├── meta_api_demo.py
│ ├── multi_provider_demo.py
│ ├── ollama_integration_demo.py
│ ├── prompt_templates_demo.py
│ ├── python_sandbox_demo.py
│ ├── rag_example.py
│ ├── research_workflow_demo.py
│ ├── sample
│ │ ├── article.txt
│ │ ├── backprop_paper.pdf
│ │ ├── buffett.pdf
│ │ ├── contract_link.txt
│ │ ├── legal_contract.txt
│ │ ├── medical_case.txt
│ │ ├── northwind.db
│ │ ├── research_paper.txt
│ │ ├── sample_data.json
│ │ └── text_classification_samples
│ │ ├── email_classification.txt
│ │ ├── news_samples.txt
│ │ ├── product_reviews.txt
│ │ └── support_tickets.txt
│ ├── sample_docs
│ │ └── downloaded
│ │ └── attention_is_all_you_need.pdf
│ ├── sentiment_analysis_demo.py
│ ├── simple_completion_demo.py
│ ├── single_shot_synthesis_demo.py
│ ├── smart_browser_demo.py
│ ├── sql_database_demo.py
│ ├── sse_client_demo.py
│ ├── test_code_extraction.py
│ ├── test_content_detection.py
│ ├── test_ollama.py
│ ├── text_classification_demo.py
│ ├── text_redline_demo.py
│ ├── tool_composition_examples.py
│ ├── tournament_code_demo.py
│ ├── tournament_text_demo.py
│ ├── unified_memory_system_demo.py
│ ├── vector_search_demo.py
│ ├── web_automation_instruction_packs.py
│ └── workflow_delegation_demo.py
├── LICENSE
├── list_models.py
├── marqo_index_config.json.example
├── mcp_protocol_schema_2025-03-25_version.json
├── mcp_python_lib_docs.md
├── mcp_tool_context_estimator.py
├── model_preferences.py
├── pyproject.toml
├── quick_test.py
├── README.md
├── resource_annotations.py
├── run_all_demo_scripts_and_check_for_errors.py
├── storage
│ └── smart_browser_internal
│ ├── locator_cache.db
│ ├── readability.js
│ └── storage_state.enc
├── test_client.py
├── test_connection.py
├── TEST_README.md
├── test_sse_client.py
├── test_stdio_client.py
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── integration
│ │ ├── __init__.py
│ │ └── test_server.py
│ ├── manual
│ │ ├── test_extraction_advanced.py
│ │ └── test_extraction.py
│ └── unit
│ ├── __init__.py
│ ├── test_cache.py
│ ├── test_providers.py
│ └── test_tools.py
├── TODO.md
├── tool_annotations.py
├── tools_list.json
├── ultimate_mcp_banner.webp
├── ultimate_mcp_logo.webp
├── ultimate_mcp_server
│ ├── __init__.py
│ ├── __main__.py
│ ├── cli
│ │ ├── __init__.py
│ │ ├── __main__.py
│ │ ├── commands.py
│ │ ├── helpers.py
│ │ └── typer_cli.py
│ ├── clients
│ │ ├── __init__.py
│ │ ├── completion_client.py
│ │ └── rag_client.py
│ ├── config
│ │ └── examples
│ │ └── filesystem_config.yaml
│ ├── config.py
│ ├── constants.py
│ ├── core
│ │ ├── __init__.py
│ │ ├── evaluation
│ │ │ ├── base.py
│ │ │ └── evaluators.py
│ │ ├── providers
│ │ │ ├── __init__.py
│ │ │ ├── anthropic.py
│ │ │ ├── base.py
│ │ │ ├── deepseek.py
│ │ │ ├── gemini.py
│ │ │ ├── grok.py
│ │ │ ├── ollama.py
│ │ │ ├── openai.py
│ │ │ └── openrouter.py
│ │ ├── server.py
│ │ ├── state_store.py
│ │ ├── tournaments
│ │ │ ├── manager.py
│ │ │ ├── tasks.py
│ │ │ └── utils.py
│ │ └── ums_api
│ │ ├── __init__.py
│ │ ├── ums_database.py
│ │ ├── ums_endpoints.py
│ │ ├── ums_models.py
│ │ └── ums_services.py
│ ├── exceptions.py
│ ├── graceful_shutdown.py
│ ├── services
│ │ ├── __init__.py
│ │ ├── analytics
│ │ │ ├── __init__.py
│ │ │ ├── metrics.py
│ │ │ └── reporting.py
│ │ ├── cache
│ │ │ ├── __init__.py
│ │ │ ├── cache_service.py
│ │ │ ├── persistence.py
│ │ │ ├── strategies.py
│ │ │ └── utils.py
│ │ ├── cache.py
│ │ ├── document.py
│ │ ├── knowledge_base
│ │ │ ├── __init__.py
│ │ │ ├── feedback.py
│ │ │ ├── manager.py
│ │ │ ├── rag_engine.py
│ │ │ ├── retriever.py
│ │ │ └── utils.py
│ │ ├── prompts
│ │ │ ├── __init__.py
│ │ │ ├── repository.py
│ │ │ └── templates.py
│ │ ├── prompts.py
│ │ └── vector
│ │ ├── __init__.py
│ │ ├── embeddings.py
│ │ └── vector_service.py
│ ├── tool_token_counter.py
│ ├── tools
│ │ ├── __init__.py
│ │ ├── audio_transcription.py
│ │ ├── base.py
│ │ ├── completion.py
│ │ ├── docstring_refiner.py
│ │ ├── document_conversion_and_processing.py
│ │ ├── enhanced-ums-lookbook.html
│ │ ├── entity_relation_graph.py
│ │ ├── excel_spreadsheet_automation.py
│ │ ├── extraction.py
│ │ ├── filesystem.py
│ │ ├── html_to_markdown.py
│ │ ├── local_text_tools.py
│ │ ├── marqo_fused_search.py
│ │ ├── meta_api_tool.py
│ │ ├── ocr_tools.py
│ │ ├── optimization.py
│ │ ├── provider.py
│ │ ├── pyodide_boot_template.html
│ │ ├── python_sandbox.py
│ │ ├── rag.py
│ │ ├── redline-compiled.css
│ │ ├── sentiment_analysis.py
│ │ ├── single_shot_synthesis.py
│ │ ├── smart_browser.py
│ │ ├── sql_databases.py
│ │ ├── text_classification.py
│ │ ├── text_redline_tools.py
│ │ ├── tournament.py
│ │ ├── ums_explorer.html
│ │ └── unified_memory_system.py
│ ├── utils
│ │ ├── __init__.py
│ │ ├── async_utils.py
│ │ ├── display.py
│ │ ├── logging
│ │ │ ├── __init__.py
│ │ │ ├── console.py
│ │ │ ├── emojis.py
│ │ │ ├── formatter.py
│ │ │ ├── logger.py
│ │ │ ├── panels.py
│ │ │ ├── progress.py
│ │ │ └── themes.py
│ │ ├── parse_yaml.py
│ │ ├── parsing.py
│ │ ├── security.py
│ │ └── text.py
│ └── working_memory_api.py
├── unified_memory_system_technical_analysis.md
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/mcp_protocol_schema_2025-03-25_version.json:
--------------------------------------------------------------------------------
```json
{
"$schema": "http://json-schema.org/draft-07/schema#",
"definitions": {
"Annotations": {
"description": "Optional annotations for the client. The client can use annotations to inform how objects are used or displayed",
"properties": {
"audience": {
"description": "Describes who the intended customer of this object or data is.\n\nIt can include multiple entries to indicate content useful for multiple audiences (e.g., `[\"user\", \"assistant\"]`).",
"items": {
"$ref": "#/definitions/Role"
},
"type": "array"
},
"priority": {
"description": "Describes how important this data is for operating the server.\n\nA value of 1 means \"most important,\" and indicates that the data is\neffectively required, while 0 means \"least important,\" and indicates that\nthe data is entirely optional.",
"maximum": 1,
"minimum": 0,
"type": "number"
}
},
"type": "object"
},
"AudioContent": {
"description": "Audio provided to or from an LLM.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"data": {
"description": "The base64-encoded audio data.",
"format": "byte",
"type": "string"
},
"mimeType": {
"description": "The MIME type of the audio. Different providers may support different audio types.",
"type": "string"
},
"type": {
"const": "audio",
"type": "string"
}
},
"required": [
"data",
"mimeType",
"type"
],
"type": "object"
},
"BlobResourceContents": {
"properties": {
"blob": {
"description": "A base64-encoded string representing the binary data of the item.",
"format": "byte",
"type": "string"
},
"mimeType": {
"description": "The MIME type of this resource, if known.",
"type": "string"
},
"uri": {
"description": "The URI of this resource.",
"format": "uri",
"type": "string"
}
},
"required": [
"blob",
"uri"
],
"type": "object"
},
"CallToolRequest": {
"description": "Used by the client to invoke a tool provided by the server.",
"properties": {
"method": {
"const": "tools/call",
"type": "string"
},
"params": {
"properties": {
"arguments": {
"additionalProperties": {},
"type": "object"
},
"name": {
"type": "string"
}
},
"required": [
"name"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"CallToolResult": {
"description": "The server's response to a tool call.\n\nAny errors that originate from the tool SHOULD be reported inside the result\nobject, with `isError` set to true, _not_ as an MCP protocol-level error\nresponse. Otherwise, the LLM would not be able to see that an error occurred\nand self-correct.\n\nHowever, any errors in _finding_ the tool, an error indicating that the\nserver does not support tool calls, or any other exceptional conditions,\nshould be reported as an MCP error response.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"content": {
"items": {
"anyOf": [
{
"$ref": "#/definitions/TextContent"
},
{
"$ref": "#/definitions/ImageContent"
},
{
"$ref": "#/definitions/AudioContent"
},
{
"$ref": "#/definitions/EmbeddedResource"
}
]
},
"type": "array"
},
"isError": {
"description": "Whether the tool call ended in an error.\n\nIf not set, this is assumed to be false (the call was successful).",
"type": "boolean"
}
},
"required": [
"content"
],
"type": "object"
},
"CancelledNotification": {
"description": "This notification can be sent by either side to indicate that it is cancelling a previously-issued request.\n\nThe request SHOULD still be in-flight, but due to communication latency, it is always possible that this notification MAY arrive after the request has already finished.\n\nThis notification indicates that the result will be unused, so any associated processing SHOULD cease.\n\nA client MUST NOT attempt to cancel its `initialize` request.",
"properties": {
"method": {
"const": "notifications/cancelled",
"type": "string"
},
"params": {
"properties": {
"reason": {
"description": "An optional string describing the reason for the cancellation. This MAY be logged or presented to the user.",
"type": "string"
},
"requestId": {
"$ref": "#/definitions/RequestId",
"description": "The ID of the request to cancel.\n\nThis MUST correspond to the ID of a request previously issued in the same direction."
}
},
"required": [
"requestId"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"ClientCapabilities": {
"description": "Capabilities a client may support. Known capabilities are defined here, in this schema, but this is not a closed set: any client can define its own, additional capabilities.",
"properties": {
"experimental": {
"additionalProperties": {
"additionalProperties": true,
"properties": {},
"type": "object"
},
"description": "Experimental, non-standard capabilities that the client supports.",
"type": "object"
},
"roots": {
"description": "Present if the client supports listing roots.",
"properties": {
"listChanged": {
"description": "Whether the client supports notifications for changes to the roots list.",
"type": "boolean"
}
},
"type": "object"
},
"sampling": {
"additionalProperties": true,
"description": "Present if the client supports sampling from an LLM.",
"properties": {},
"type": "object"
}
},
"type": "object"
},
"ClientNotification": {
"anyOf": [
{
"$ref": "#/definitions/CancelledNotification"
},
{
"$ref": "#/definitions/InitializedNotification"
},
{
"$ref": "#/definitions/ProgressNotification"
},
{
"$ref": "#/definitions/RootsListChangedNotification"
}
]
},
"ClientRequest": {
"anyOf": [
{
"$ref": "#/definitions/InitializeRequest"
},
{
"$ref": "#/definitions/PingRequest"
},
{
"$ref": "#/definitions/ListResourcesRequest"
},
{
"$ref": "#/definitions/ReadResourceRequest"
},
{
"$ref": "#/definitions/SubscribeRequest"
},
{
"$ref": "#/definitions/UnsubscribeRequest"
},
{
"$ref": "#/definitions/ListPromptsRequest"
},
{
"$ref": "#/definitions/GetPromptRequest"
},
{
"$ref": "#/definitions/ListToolsRequest"
},
{
"$ref": "#/definitions/CallToolRequest"
},
{
"$ref": "#/definitions/SetLevelRequest"
},
{
"$ref": "#/definitions/CompleteRequest"
}
]
},
"ClientResult": {
"anyOf": [
{
"$ref": "#/definitions/Result"
},
{
"$ref": "#/definitions/CreateMessageResult"
},
{
"$ref": "#/definitions/ListRootsResult"
}
]
},
"CompleteRequest": {
"description": "A request from the client to the server, to ask for completion options.",
"properties": {
"method": {
"const": "completion/complete",
"type": "string"
},
"params": {
"properties": {
"argument": {
"description": "The argument's information",
"properties": {
"name": {
"description": "The name of the argument",
"type": "string"
},
"value": {
"description": "The value of the argument to use for completion matching.",
"type": "string"
}
},
"required": [
"name",
"value"
],
"type": "object"
},
"ref": {
"anyOf": [
{
"$ref": "#/definitions/PromptReference"
},
{
"$ref": "#/definitions/ResourceReference"
}
]
}
},
"required": [
"argument",
"ref"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"CompleteResult": {
"description": "The server's response to a completion/complete request",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"completion": {
"properties": {
"hasMore": {
"description": "Indicates whether there are additional completion options beyond those provided in the current response, even if the exact total is unknown.",
"type": "boolean"
},
"total": {
"description": "The total number of completion options available. This can exceed the number of values actually sent in the response.",
"type": "integer"
},
"values": {
"description": "An array of completion values. Must not exceed 100 items.",
"items": {
"type": "string"
},
"type": "array"
}
},
"required": [
"values"
],
"type": "object"
}
},
"required": [
"completion"
],
"type": "object"
},
"CreateMessageRequest": {
"description": "A request from the server to sample an LLM via the client. The client has full discretion over which model to select. The client should also inform the user before beginning sampling, to allow them to inspect the request (human in the loop) and decide whether to approve it.",
"properties": {
"method": {
"const": "sampling/createMessage",
"type": "string"
},
"params": {
"properties": {
"includeContext": {
"description": "A request to include context from one or more MCP servers (including the caller), to be attached to the prompt. The client MAY ignore this request.",
"enum": [
"allServers",
"none",
"thisServer"
],
"type": "string"
},
"maxTokens": {
"description": "The maximum number of tokens to sample, as requested by the server. The client MAY choose to sample fewer tokens than requested.",
"type": "integer"
},
"messages": {
"items": {
"$ref": "#/definitions/SamplingMessage"
},
"type": "array"
},
"metadata": {
"additionalProperties": true,
"description": "Optional metadata to pass through to the LLM provider. The format of this metadata is provider-specific.",
"properties": {},
"type": "object"
},
"modelPreferences": {
"$ref": "#/definitions/ModelPreferences",
"description": "The server's preferences for which model to select. The client MAY ignore these preferences."
},
"stopSequences": {
"items": {
"type": "string"
},
"type": "array"
},
"systemPrompt": {
"description": "An optional system prompt the server wants to use for sampling. The client MAY modify or omit this prompt.",
"type": "string"
},
"temperature": {
"type": "number"
}
},
"required": [
"maxTokens",
"messages"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"CreateMessageResult": {
"description": "The client's response to a sampling/create_message request from the server. The client should inform the user before returning the sampled message, to allow them to inspect the response (human in the loop) and decide whether to allow the server to see it.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"content": {
"anyOf": [
{
"$ref": "#/definitions/TextContent"
},
{
"$ref": "#/definitions/ImageContent"
},
{
"$ref": "#/definitions/AudioContent"
}
]
},
"model": {
"description": "The name of the model that generated the message.",
"type": "string"
},
"role": {
"$ref": "#/definitions/Role"
},
"stopReason": {
"description": "The reason why sampling stopped, if known.",
"type": "string"
}
},
"required": [
"content",
"model",
"role"
],
"type": "object"
},
"Cursor": {
"description": "An opaque token used to represent a cursor for pagination.",
"type": "string"
},
"EmbeddedResource": {
"description": "The contents of a resource, embedded into a prompt or tool call result.\n\nIt is up to the client how best to render embedded resources for the benefit\nof the LLM and/or the user.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"resource": {
"anyOf": [
{
"$ref": "#/definitions/TextResourceContents"
},
{
"$ref": "#/definitions/BlobResourceContents"
}
]
},
"type": {
"const": "resource",
"type": "string"
}
},
"required": [
"resource",
"type"
],
"type": "object"
},
"EmptyResult": {
"$ref": "#/definitions/Result"
},
"GetPromptRequest": {
"description": "Used by the client to get a prompt provided by the server.",
"properties": {
"method": {
"const": "prompts/get",
"type": "string"
},
"params": {
"properties": {
"arguments": {
"additionalProperties": {
"type": "string"
},
"description": "Arguments to use for templating the prompt.",
"type": "object"
},
"name": {
"description": "The name of the prompt or prompt template.",
"type": "string"
}
},
"required": [
"name"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"GetPromptResult": {
"description": "The server's response to a prompts/get request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"description": {
"description": "An optional description for the prompt.",
"type": "string"
},
"messages": {
"items": {
"$ref": "#/definitions/PromptMessage"
},
"type": "array"
}
},
"required": [
"messages"
],
"type": "object"
},
"ImageContent": {
"description": "An image provided to or from an LLM.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"data": {
"description": "The base64-encoded image data.",
"format": "byte",
"type": "string"
},
"mimeType": {
"description": "The MIME type of the image. Different providers may support different image types.",
"type": "string"
},
"type": {
"const": "image",
"type": "string"
}
},
"required": [
"data",
"mimeType",
"type"
],
"type": "object"
},
"Implementation": {
"description": "Describes the name and version of an MCP implementation.",
"properties": {
"name": {
"type": "string"
},
"version": {
"type": "string"
}
},
"required": [
"name",
"version"
],
"type": "object"
},
"InitializeRequest": {
"description": "This request is sent from the client to the server when it first connects, asking it to begin initialization.",
"properties": {
"method": {
"const": "initialize",
"type": "string"
},
"params": {
"properties": {
"capabilities": {
"$ref": "#/definitions/ClientCapabilities"
},
"clientInfo": {
"$ref": "#/definitions/Implementation"
},
"protocolVersion": {
"description": "The latest version of the Model Context Protocol that the client supports. The client MAY decide to support older versions as well.",
"type": "string"
}
},
"required": [
"capabilities",
"clientInfo",
"protocolVersion"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"InitializeResult": {
"description": "After receiving an initialize request from the client, the server sends this response.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"capabilities": {
"$ref": "#/definitions/ServerCapabilities"
},
"instructions": {
"description": "Instructions describing how to use the server and its features.\n\nThis can be used by clients to improve the LLM's understanding of available tools, resources, etc. It can be thought of like a \"hint\" to the model. For example, this information MAY be added to the system prompt.",
"type": "string"
},
"protocolVersion": {
"description": "The version of the Model Context Protocol that the server wants to use. This may not match the version that the client requested. If the client cannot support this version, it MUST disconnect.",
"type": "string"
},
"serverInfo": {
"$ref": "#/definitions/Implementation"
}
},
"required": [
"capabilities",
"protocolVersion",
"serverInfo"
],
"type": "object"
},
"InitializedNotification": {
"description": "This notification is sent from the client to the server after initialization has finished.",
"properties": {
"method": {
"const": "notifications/initialized",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"JSONRPCBatchRequest": {
"description": "A JSON-RPC batch request, as described in https://www.jsonrpc.org/specification#batch.",
"items": {
"anyOf": [
{
"$ref": "#/definitions/JSONRPCRequest"
},
{
"$ref": "#/definitions/JSONRPCNotification"
}
]
},
"type": "array"
},
"JSONRPCBatchResponse": {
"description": "A JSON-RPC batch response, as described in https://www.jsonrpc.org/specification#batch.",
"items": {
"anyOf": [
{
"$ref": "#/definitions/JSONRPCResponse"
},
{
"$ref": "#/definitions/JSONRPCError"
}
]
},
"type": "array"
},
"JSONRPCError": {
"description": "A response to a request that indicates an error occurred.",
"properties": {
"error": {
"properties": {
"code": {
"description": "The error type that occurred.",
"type": "integer"
},
"data": {
"description": "Additional information about the error. The value of this member is defined by the sender (e.g. detailed error information, nested errors etc.)."
},
"message": {
"description": "A short description of the error. The message SHOULD be limited to a concise single sentence.",
"type": "string"
}
},
"required": [
"code",
"message"
],
"type": "object"
},
"id": {
"$ref": "#/definitions/RequestId"
},
"jsonrpc": {
"const": "2.0",
"type": "string"
}
},
"required": [
"error",
"id",
"jsonrpc"
],
"type": "object"
},
"JSONRPCMessage": {
"anyOf": [
{
"$ref": "#/definitions/JSONRPCRequest"
},
{
"$ref": "#/definitions/JSONRPCNotification"
},
{
"description": "A JSON-RPC batch request, as described in https://www.jsonrpc.org/specification#batch.",
"items": {
"anyOf": [
{
"$ref": "#/definitions/JSONRPCRequest"
},
{
"$ref": "#/definitions/JSONRPCNotification"
}
]
},
"type": "array"
},
{
"$ref": "#/definitions/JSONRPCResponse"
},
{
"$ref": "#/definitions/JSONRPCError"
},
{
"description": "A JSON-RPC batch response, as described in https://www.jsonrpc.org/specification#batch.",
"items": {
"anyOf": [
{
"$ref": "#/definitions/JSONRPCResponse"
},
{
"$ref": "#/definitions/JSONRPCError"
}
]
},
"type": "array"
}
],
"description": "Refers to any valid JSON-RPC object that can be decoded off the wire, or encoded to be sent."
},
"JSONRPCNotification": {
"description": "A notification which does not expect a response.",
"properties": {
"jsonrpc": {
"const": "2.0",
"type": "string"
},
"method": {
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"jsonrpc",
"method"
],
"type": "object"
},
"JSONRPCRequest": {
"description": "A request that expects a response.",
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"jsonrpc": {
"const": "2.0",
"type": "string"
},
"method": {
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"properties": {
"progressToken": {
"$ref": "#/definitions/ProgressToken",
"description": "If specified, the caller is requesting out-of-band progress notifications for this request (as represented by notifications/progress). The value of this parameter is an opaque token that will be attached to any subsequent notifications. The receiver is not obligated to provide these notifications."
}
},
"type": "object"
}
},
"type": "object"
}
},
"required": [
"id",
"jsonrpc",
"method"
],
"type": "object"
},
"JSONRPCResponse": {
"description": "A successful (non-error) response to a request.",
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"jsonrpc": {
"const": "2.0",
"type": "string"
},
"result": {
"$ref": "#/definitions/Result"
}
},
"required": [
"id",
"jsonrpc",
"result"
],
"type": "object"
},
"ListPromptsRequest": {
"description": "Sent from the client to request a list of prompts and prompt templates the server has.",
"properties": {
"method": {
"const": "prompts/list",
"type": "string"
},
"params": {
"properties": {
"cursor": {
"description": "An opaque token representing the current pagination position.\nIf provided, the server should return results starting after this cursor.",
"type": "string"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ListPromptsResult": {
"description": "The server's response to a prompts/list request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"nextCursor": {
"description": "An opaque token representing the pagination position after the last returned result.\nIf present, there may be more results available.",
"type": "string"
},
"prompts": {
"items": {
"$ref": "#/definitions/Prompt"
},
"type": "array"
}
},
"required": [
"prompts"
],
"type": "object"
},
"ListResourceTemplatesRequest": {
"description": "Sent from the client to request a list of resource templates the server has.",
"properties": {
"method": {
"const": "resources/templates/list",
"type": "string"
},
"params": {
"properties": {
"cursor": {
"description": "An opaque token representing the current pagination position.\nIf provided, the server should return results starting after this cursor.",
"type": "string"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ListResourceTemplatesResult": {
"description": "The server's response to a resources/templates/list request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"nextCursor": {
"description": "An opaque token representing the pagination position after the last returned result.\nIf present, there may be more results available.",
"type": "string"
},
"resourceTemplates": {
"items": {
"$ref": "#/definitions/ResourceTemplate"
},
"type": "array"
}
},
"required": [
"resourceTemplates"
],
"type": "object"
},
"ListResourcesRequest": {
"description": "Sent from the client to request a list of resources the server has.",
"properties": {
"method": {
"const": "resources/list",
"type": "string"
},
"params": {
"properties": {
"cursor": {
"description": "An opaque token representing the current pagination position.\nIf provided, the server should return results starting after this cursor.",
"type": "string"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ListResourcesResult": {
"description": "The server's response to a resources/list request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"nextCursor": {
"description": "An opaque token representing the pagination position after the last returned result.\nIf present, there may be more results available.",
"type": "string"
},
"resources": {
"items": {
"$ref": "#/definitions/Resource"
},
"type": "array"
}
},
"required": [
"resources"
],
"type": "object"
},
"ListRootsRequest": {
"description": "Sent from the server to request a list of root URIs from the client. Roots allow\nservers to ask for specific directories or files to operate on. A common example\nfor roots is providing a set of repositories or directories a server should operate\non.\n\nThis request is typically used when the server needs to understand the file system\nstructure or access specific locations that the client has permission to read from.",
"properties": {
"method": {
"const": "roots/list",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"properties": {
"progressToken": {
"$ref": "#/definitions/ProgressToken",
"description": "If specified, the caller is requesting out-of-band progress notifications for this request (as represented by notifications/progress). The value of this parameter is an opaque token that will be attached to any subsequent notifications. The receiver is not obligated to provide these notifications."
}
},
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ListRootsResult": {
"description": "The client's response to a roots/list request from the server.\nThis result contains an array of Root objects, each representing a root directory\nor file that the server can operate on.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"roots": {
"items": {
"$ref": "#/definitions/Root"
},
"type": "array"
}
},
"required": [
"roots"
],
"type": "object"
},
"ListToolsRequest": {
"description": "Sent from the client to request a list of tools the server has.",
"properties": {
"method": {
"const": "tools/list",
"type": "string"
},
"params": {
"properties": {
"cursor": {
"description": "An opaque token representing the current pagination position.\nIf provided, the server should return results starting after this cursor.",
"type": "string"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ListToolsResult": {
"description": "The server's response to a tools/list request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"nextCursor": {
"description": "An opaque token representing the pagination position after the last returned result.\nIf present, there may be more results available.",
"type": "string"
},
"tools": {
"items": {
"$ref": "#/definitions/Tool"
},
"type": "array"
}
},
"required": [
"tools"
],
"type": "object"
},
"LoggingLevel": {
"description": "The severity of a log message.\n\nThese map to syslog message severities, as specified in RFC-5424:\nhttps://datatracker.ietf.org/doc/html/rfc5424#section-6.2.1",
"enum": [
"alert",
"critical",
"debug",
"emergency",
"error",
"info",
"notice",
"warning"
],
"type": "string"
},
"LoggingMessageNotification": {
"description": "Notification of a log message passed from server to client. If no logging/setLevel request has been sent from the client, the server MAY decide which messages to send automatically.",
"properties": {
"method": {
"const": "notifications/message",
"type": "string"
},
"params": {
"properties": {
"data": {
"description": "The data to be logged, such as a string message or an object. Any JSON serializable type is allowed here."
},
"level": {
"$ref": "#/definitions/LoggingLevel",
"description": "The severity of this log message."
},
"logger": {
"description": "An optional name of the logger issuing this message.",
"type": "string"
}
},
"required": [
"data",
"level"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"ModelHint": {
"description": "Hints to use for model selection.\n\nKeys not declared here are currently left unspecified by the spec and are up\nto the client to interpret.",
"properties": {
"name": {
"description": "A hint for a model name.\n\nThe client SHOULD treat this as a substring of a model name; for example:\n - `claude-3-5-sonnet` should match `claude-3-5-sonnet-20241022`\n - `sonnet` should match `claude-3-5-sonnet-20241022`, `claude-3-sonnet-20240229`, etc.\n - `claude` should match any Claude model\n\nThe client MAY also map the string to a different provider's model name or a different model family, as long as it fills a similar niche; for example:\n - `gemini-1.5-flash` could match `claude-3-5-haiku-20241022`",
"type": "string"
}
},
"type": "object"
},
"ModelPreferences": {
"description": "The server's preferences for model selection, requested of the client during sampling.\n\nBecause LLMs can vary along multiple dimensions, choosing the \"best\" model is\nrarely straightforward. Different models excel in different areas—some are\nfaster but less capable, others are more capable but more expensive, and so\non. This interface allows servers to express their priorities across multiple\ndimensions to help clients make an appropriate selection for their use case.\n\nThese preferences are always advisory. The client MAY ignore them. It is also\nup to the client to decide how to interpret these preferences and how to\nbalance them against other considerations.",
"properties": {
"costPriority": {
"description": "How much to prioritize cost when selecting a model. A value of 0 means cost\nis not important, while a value of 1 means cost is the most important\nfactor.",
"maximum": 1,
"minimum": 0,
"type": "number"
},
"hints": {
"description": "Optional hints to use for model selection.\n\nIf multiple hints are specified, the client MUST evaluate them in order\n(such that the first match is taken).\n\nThe client SHOULD prioritize these hints over the numeric priorities, but\nMAY still use the priorities to select from ambiguous matches.",
"items": {
"$ref": "#/definitions/ModelHint"
},
"type": "array"
},
"intelligencePriority": {
"description": "How much to prioritize intelligence and capabilities when selecting a\nmodel. A value of 0 means intelligence is not important, while a value of 1\nmeans intelligence is the most important factor.",
"maximum": 1,
"minimum": 0,
"type": "number"
},
"speedPriority": {
"description": "How much to prioritize sampling speed (latency) when selecting a model. A\nvalue of 0 means speed is not important, while a value of 1 means speed is\nthe most important factor.",
"maximum": 1,
"minimum": 0,
"type": "number"
}
},
"type": "object"
},
"Notification": {
"properties": {
"method": {
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"PaginatedRequest": {
"properties": {
"method": {
"type": "string"
},
"params": {
"properties": {
"cursor": {
"description": "An opaque token representing the current pagination position.\nIf provided, the server should return results starting after this cursor.",
"type": "string"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"PaginatedResult": {
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"nextCursor": {
"description": "An opaque token representing the pagination position after the last returned result.\nIf present, there may be more results available.",
"type": "string"
}
},
"type": "object"
},
"PingRequest": {
"description": "A ping, issued by either the server or the client, to check that the other party is still alive. The receiver must promptly respond, or else may be disconnected.",
"properties": {
"method": {
"const": "ping",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"properties": {
"progressToken": {
"$ref": "#/definitions/ProgressToken",
"description": "If specified, the caller is requesting out-of-band progress notifications for this request (as represented by notifications/progress). The value of this parameter is an opaque token that will be attached to any subsequent notifications. The receiver is not obligated to provide these notifications."
}
},
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ProgressNotification": {
"description": "An out-of-band notification used to inform the receiver of a progress update for a long-running request.",
"properties": {
"method": {
"const": "notifications/progress",
"type": "string"
},
"params": {
"properties": {
"message": {
"description": "An optional message describing the current progress.",
"type": "string"
},
"progress": {
"description": "The progress thus far. This should increase every time progress is made, even if the total is unknown.",
"type": "number"
},
"progressToken": {
"$ref": "#/definitions/ProgressToken",
"description": "The progress token which was given in the initial request, used to associate this notification with the request that is proceeding."
},
"total": {
"description": "Total number of items to process (or total progress required), if known.",
"type": "number"
}
},
"required": [
"progress",
"progressToken"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"ProgressToken": {
"description": "A progress token, used to associate progress notifications with the original request.",
"type": [
"string",
"integer"
]
},
"Prompt": {
"description": "A prompt or prompt template that the server offers.",
"properties": {
"arguments": {
"description": "A list of arguments to use for templating the prompt.",
"items": {
"$ref": "#/definitions/PromptArgument"
},
"type": "array"
},
"description": {
"description": "An optional description of what this prompt provides",
"type": "string"
},
"name": {
"description": "The name of the prompt or prompt template.",
"type": "string"
}
},
"required": [
"name"
],
"type": "object"
},
"PromptArgument": {
"description": "Describes an argument that a prompt can accept.",
"properties": {
"description": {
"description": "A human-readable description of the argument.",
"type": "string"
},
"name": {
"description": "The name of the argument.",
"type": "string"
},
"required": {
"description": "Whether this argument must be provided.",
"type": "boolean"
}
},
"required": [
"name"
],
"type": "object"
},
"PromptListChangedNotification": {
"description": "An optional notification from the server to the client, informing it that the list of prompts it offers has changed. This may be issued by servers without any previous subscription from the client.",
"properties": {
"method": {
"const": "notifications/prompts/list_changed",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"PromptMessage": {
"description": "Describes a message returned as part of a prompt.\n\nThis is similar to `SamplingMessage`, but also supports the embedding of\nresources from the MCP server.",
"properties": {
"content": {
"anyOf": [
{
"$ref": "#/definitions/TextContent"
},
{
"$ref": "#/definitions/ImageContent"
},
{
"$ref": "#/definitions/AudioContent"
},
{
"$ref": "#/definitions/EmbeddedResource"
}
]
},
"role": {
"$ref": "#/definitions/Role"
}
},
"required": [
"content",
"role"
],
"type": "object"
},
"PromptReference": {
"description": "Identifies a prompt.",
"properties": {
"name": {
"description": "The name of the prompt or prompt template",
"type": "string"
},
"type": {
"const": "ref/prompt",
"type": "string"
}
},
"required": [
"name",
"type"
],
"type": "object"
},
"ReadResourceRequest": {
"description": "Sent from the client to the server, to read a specific resource URI.",
"properties": {
"method": {
"const": "resources/read",
"type": "string"
},
"params": {
"properties": {
"uri": {
"description": "The URI of the resource to read. The URI can use any protocol; it is up to the server how to interpret it.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"ReadResourceResult": {
"description": "The server's response to a resources/read request from the client.",
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
},
"contents": {
"items": {
"anyOf": [
{
"$ref": "#/definitions/TextResourceContents"
},
{
"$ref": "#/definitions/BlobResourceContents"
}
]
},
"type": "array"
}
},
"required": [
"contents"
],
"type": "object"
},
"Request": {
"properties": {
"method": {
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"properties": {
"progressToken": {
"$ref": "#/definitions/ProgressToken",
"description": "If specified, the caller is requesting out-of-band progress notifications for this request (as represented by notifications/progress). The value of this parameter is an opaque token that will be attached to any subsequent notifications. The receiver is not obligated to provide these notifications."
}
},
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"RequestId": {
"description": "A uniquely identifying ID for a request in JSON-RPC.",
"type": [
"string",
"integer"
]
},
"Resource": {
"description": "A known resource that the server is capable of reading.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"description": {
"description": "A description of what this resource represents.\n\nThis can be used by clients to improve the LLM's understanding of available resources. It can be thought of like a \"hint\" to the model.",
"type": "string"
},
"mimeType": {
"description": "The MIME type of this resource, if known.",
"type": "string"
},
"name": {
"description": "A human-readable name for this resource.\n\nThis can be used by clients to populate UI elements.",
"type": "string"
},
"size": {
"description": "The size of the raw resource content, in bytes (i.e., before base64 encoding or any tokenization), if known.\n\nThis can be used by Hosts to display file sizes and estimate context window usage.",
"type": "integer"
},
"uri": {
"description": "The URI of this resource.",
"format": "uri",
"type": "string"
}
},
"required": [
"name",
"uri"
],
"type": "object"
},
"ResourceContents": {
"description": "The contents of a specific resource or sub-resource.",
"properties": {
"mimeType": {
"description": "The MIME type of this resource, if known.",
"type": "string"
},
"uri": {
"description": "The URI of this resource.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
},
"ResourceListChangedNotification": {
"description": "An optional notification from the server to the client, informing it that the list of resources it can read from has changed. This may be issued by servers without any previous subscription from the client.",
"properties": {
"method": {
"const": "notifications/resources/list_changed",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"ResourceReference": {
"description": "A reference to a resource or resource template definition.",
"properties": {
"type": {
"const": "ref/resource",
"type": "string"
},
"uri": {
"description": "The URI or URI template of the resource.",
"format": "uri-template",
"type": "string"
}
},
"required": [
"type",
"uri"
],
"type": "object"
},
"ResourceTemplate": {
"description": "A template description for resources available on the server.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"description": {
"description": "A description of what this template is for.\n\nThis can be used by clients to improve the LLM's understanding of available resources. It can be thought of like a \"hint\" to the model.",
"type": "string"
},
"mimeType": {
"description": "The MIME type for all resources that match this template. This should only be included if all resources matching this template have the same type.",
"type": "string"
},
"name": {
"description": "A human-readable name for the type of resource this template refers to.\n\nThis can be used by clients to populate UI elements.",
"type": "string"
},
"uriTemplate": {
"description": "A URI template (according to RFC 6570) that can be used to construct resource URIs.",
"format": "uri-template",
"type": "string"
}
},
"required": [
"name",
"uriTemplate"
],
"type": "object"
},
"ResourceUpdatedNotification": {
"description": "A notification from the server to the client, informing it that a resource has changed and may need to be read again. This should only be sent if the client previously sent a resources/subscribe request.",
"properties": {
"method": {
"const": "notifications/resources/updated",
"type": "string"
},
"params": {
"properties": {
"uri": {
"description": "The URI of the resource that has been updated. This might be a sub-resource of the one that the client actually subscribed to.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"Result": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This result property is reserved by the protocol to allow clients and servers to attach additional metadata to their responses.",
"type": "object"
}
},
"type": "object"
},
"Role": {
"description": "The sender or recipient of messages and data in a conversation.",
"enum": [
"assistant",
"user"
],
"type": "string"
},
"Root": {
"description": "Represents a root directory or file that the server can operate on.",
"properties": {
"name": {
"description": "An optional name for the root. This can be used to provide a human-readable\nidentifier for the root, which may be useful for display purposes or for\nreferencing the root in other parts of the application.",
"type": "string"
},
"uri": {
"description": "The URI identifying the root. This *must* start with file:// for now.\nThis restriction may be relaxed in future versions of the protocol to allow\nother URI schemes.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
},
"RootsListChangedNotification": {
"description": "A notification from the client to the server, informing it that the list of roots has changed.\nThis notification should be sent whenever the client adds, removes, or modifies any root.\nThe server should then request an updated list of roots using the ListRootsRequest.",
"properties": {
"method": {
"const": "notifications/roots/list_changed",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"SamplingMessage": {
"description": "Describes a message issued to or received from an LLM API.",
"properties": {
"content": {
"anyOf": [
{
"$ref": "#/definitions/TextContent"
},
{
"$ref": "#/definitions/ImageContent"
},
{
"$ref": "#/definitions/AudioContent"
}
]
},
"role": {
"$ref": "#/definitions/Role"
}
},
"required": [
"content",
"role"
],
"type": "object"
},
"ServerCapabilities": {
"description": "Capabilities that a server may support. Known capabilities are defined here, in this schema, but this is not a closed set: any server can define its own, additional capabilities.",
"properties": {
"completions": {
"additionalProperties": true,
"description": "Present if the server supports argument autocompletion suggestions.",
"properties": {},
"type": "object"
},
"experimental": {
"additionalProperties": {
"additionalProperties": true,
"properties": {},
"type": "object"
},
"description": "Experimental, non-standard capabilities that the server supports.",
"type": "object"
},
"logging": {
"additionalProperties": true,
"description": "Present if the server supports sending log messages to the client.",
"properties": {},
"type": "object"
},
"prompts": {
"description": "Present if the server offers any prompt templates.",
"properties": {
"listChanged": {
"description": "Whether this server supports notifications for changes to the prompt list.",
"type": "boolean"
}
},
"type": "object"
},
"resources": {
"description": "Present if the server offers any resources to read.",
"properties": {
"listChanged": {
"description": "Whether this server supports notifications for changes to the resource list.",
"type": "boolean"
},
"subscribe": {
"description": "Whether this server supports subscribing to resource updates.",
"type": "boolean"
}
},
"type": "object"
},
"tools": {
"description": "Present if the server offers any tools to call.",
"properties": {
"listChanged": {
"description": "Whether this server supports notifications for changes to the tool list.",
"type": "boolean"
}
},
"type": "object"
}
},
"type": "object"
},
"ServerNotification": {
"anyOf": [
{
"$ref": "#/definitions/CancelledNotification"
},
{
"$ref": "#/definitions/ProgressNotification"
},
{
"$ref": "#/definitions/ResourceListChangedNotification"
},
{
"$ref": "#/definitions/ResourceUpdatedNotification"
},
{
"$ref": "#/definitions/PromptListChangedNotification"
},
{
"$ref": "#/definitions/ToolListChangedNotification"
},
{
"$ref": "#/definitions/LoggingMessageNotification"
}
]
},
"ServerRequest": {
"anyOf": [
{
"$ref": "#/definitions/PingRequest"
},
{
"$ref": "#/definitions/CreateMessageRequest"
},
{
"$ref": "#/definitions/ListRootsRequest"
}
]
},
"ServerResult": {
"anyOf": [
{
"$ref": "#/definitions/Result"
},
{
"$ref": "#/definitions/InitializeResult"
},
{
"$ref": "#/definitions/ListResourcesResult"
},
{
"$ref": "#/definitions/ReadResourceResult"
},
{
"$ref": "#/definitions/ListPromptsResult"
},
{
"$ref": "#/definitions/GetPromptResult"
},
{
"$ref": "#/definitions/ListToolsResult"
},
{
"$ref": "#/definitions/CallToolResult"
},
{
"$ref": "#/definitions/CompleteResult"
}
]
},
"SetLevelRequest": {
"description": "A request from the client to the server, to enable or adjust logging.",
"properties": {
"method": {
"const": "logging/setLevel",
"type": "string"
},
"params": {
"properties": {
"level": {
"$ref": "#/definitions/LoggingLevel",
"description": "The level of logging that the client wants to receive from the server. The server should send all logs at this level and higher (i.e., more severe) to the client as notifications/message."
}
},
"required": [
"level"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"SubscribeRequest": {
"description": "Sent from the client to request resources/updated notifications from the server whenever a particular resource changes.",
"properties": {
"method": {
"const": "resources/subscribe",
"type": "string"
},
"params": {
"properties": {
"uri": {
"description": "The URI of the resource to subscribe to. The URI can use any protocol; it is up to the server how to interpret it.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
},
"TextContent": {
"description": "Text provided to or from an LLM.",
"properties": {
"annotations": {
"$ref": "#/definitions/Annotations",
"description": "Optional annotations for the client."
},
"text": {
"description": "The text content of the message.",
"type": "string"
},
"type": {
"const": "text",
"type": "string"
}
},
"required": [
"text",
"type"
],
"type": "object"
},
"TextResourceContents": {
"properties": {
"mimeType": {
"description": "The MIME type of this resource, if known.",
"type": "string"
},
"text": {
"description": "The text of the item. This must only be set if the item can actually be represented as text (not binary data).",
"type": "string"
},
"uri": {
"description": "The URI of this resource.",
"format": "uri",
"type": "string"
}
},
"required": [
"text",
"uri"
],
"type": "object"
},
"Tool": {
"description": "Definition for a tool the client can call.",
"properties": {
"annotations": {
"$ref": "#/definitions/ToolAnnotations",
"description": "Optional additional tool information."
},
"description": {
"description": "A human-readable description of the tool.\n\nThis can be used by clients to improve the LLM's understanding of available tools. It can be thought of like a \"hint\" to the model.",
"type": "string"
},
"inputSchema": {
"description": "A JSON Schema object defining the expected parameters for the tool.",
"properties": {
"properties": {
"additionalProperties": {
"additionalProperties": true,
"properties": {},
"type": "object"
},
"type": "object"
},
"required": {
"items": {
"type": "string"
},
"type": "array"
},
"type": {
"const": "object",
"type": "string"
}
},
"required": [
"type"
],
"type": "object"
},
"name": {
"description": "The name of the tool.",
"type": "string"
}
},
"required": [
"inputSchema",
"name"
],
"type": "object"
},
"ToolAnnotations": {
"description": "Additional properties describing a Tool to clients.\n\nNOTE: all properties in ToolAnnotations are **hints**. \nThey are not guaranteed to provide a faithful description of \ntool behavior (including descriptive properties like `title`).\n\nClients should never make tool use decisions based on ToolAnnotations\nreceived from untrusted servers.",
"properties": {
"destructiveHint": {
"description": "If true, the tool may perform destructive updates to its environment.\nIf false, the tool performs only additive updates.\n\n(This property is meaningful only when `readOnlyHint == false`)\n\nDefault: true",
"type": "boolean"
},
"idempotentHint": {
"description": "If true, calling the tool repeatedly with the same arguments \nwill have no additional effect on the its environment.\n\n(This property is meaningful only when `readOnlyHint == false`)\n\nDefault: false",
"type": "boolean"
},
"openWorldHint": {
"description": "If true, this tool may interact with an \"open world\" of external\nentities. If false, the tool's domain of interaction is closed.\nFor example, the world of a web search tool is open, whereas that\nof a memory tool is not.\n\nDefault: true",
"type": "boolean"
},
"readOnlyHint": {
"description": "If true, the tool does not modify its environment.\n\nDefault: false",
"type": "boolean"
},
"title": {
"description": "A human-readable title for the tool.",
"type": "string"
}
},
"type": "object"
},
"ToolListChangedNotification": {
"description": "An optional notification from the server to the client, informing it that the list of tools it offers has changed. This may be issued by servers without any previous subscription from the client.",
"properties": {
"method": {
"const": "notifications/tools/list_changed",
"type": "string"
},
"params": {
"additionalProperties": {},
"properties": {
"_meta": {
"additionalProperties": {},
"description": "This parameter name is reserved by MCP to allow clients and servers to attach additional metadata to their notifications.",
"type": "object"
}
},
"type": "object"
}
},
"required": [
"method"
],
"type": "object"
},
"UnsubscribeRequest": {
"description": "Sent from the client to request cancellation of resources/updated notifications from the server. This should follow a previous resources/subscribe request.",
"properties": {
"method": {
"const": "resources/unsubscribe",
"type": "string"
},
"params": {
"properties": {
"uri": {
"description": "The URI of the resource to unsubscribe from.",
"format": "uri",
"type": "string"
}
},
"required": [
"uri"
],
"type": "object"
}
},
"required": [
"method",
"params"
],
"type": "object"
}
}
}
```
--------------------------------------------------------------------------------
/ultimate_mcp_server/tools/ocr_tools.py:
--------------------------------------------------------------------------------
```python
"""OCR Tools for Ultimate MCP Server.
This module provides tools for OCR (Optical Character Recognition) processing,
leveraging LLMs to improve the quality of extracted text from PDFs and images.
Features:
- PDF to image conversion with optimized preprocessing
- Multiple extraction methods (OCR, direct text extraction, hybrid approach)
- Intelligent text segmentation and processing for large documents
- LLM-based error correction and formatting
- Table detection and formatting
- Multi-language support
- Quality assessment with detailed metrics
- PDF structure analysis
- Batch processing with concurrency control
- Sophisticated caching for improved performance
Example usage:
```python
# Extract text from a PDF file with LLM correction
result = await client.tools.extract_text_from_pdf(
file_path="document.pdf",
extraction_method="hybrid", # Try direct text extraction first, fall back to OCR if needed
max_pages=5,
skip_pages=0,
reformat_as_markdown=True,
suppress_headers=True
)
# Process an image file with custom preprocessing
result = await client.tools.process_image_ocr(
image_path="scan.jpg",
preprocessing_options={
"denoise": True,
"threshold": "adaptive",
"deskew": True
},
ocr_language="eng+fra", # Multi-language support
assess_quality=True
)
# Enhance existing OCR text with LLM
result = await client.tools.enhance_ocr_text(
ocr_text="Text with OCK errors and broken lin- es",
reformat_as_markdown=True,
remove_headers=True
)
# Analyze PDF structure without full extraction
info = await client.tools.analyze_pdf_structure(
file_path="document.pdf",
extract_metadata=True,
extract_outline=True,
extract_fonts=True
)
# Batch process multiple PDFs
result = await client.tools.batch_process_documents(
folder_path="/path/to/documents",
file_pattern="*.pdf",
output_folder="/path/to/output",
max_concurrency=3
)
```
"""
import asyncio
import base64
import functools
import hashlib
import io
import json
import math
import os
import re
import tempfile
import time
import traceback
import uuid
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple, Union
# Try importing required libraries with fallbacks
try:
import numpy as np
HAS_NUMPY = True
except ImportError:
HAS_NUMPY = False
try:
from PIL import Image, ImageEnhance, ImageFilter
HAS_PIL = True
except ImportError:
HAS_PIL = False
try:
import cv2
HAS_CV2 = True
except ImportError:
HAS_CV2 = False
try:
import pytesseract
HAS_PYTESSERACT = True
except ImportError:
HAS_PYTESSERACT = False
try:
from pdf2image import convert_from_bytes, convert_from_path
HAS_PDF2IMAGE = True
except ImportError:
HAS_PDF2IMAGE = False
try:
import pdfplumber
HAS_PDFPLUMBER = True
except ImportError:
HAS_PDFPLUMBER = False
try:
import pymupdf # PyMuPDF
HAS_PYMUPDF = True
except ImportError:
HAS_PYMUPDF = False
# Import tools and helpers from ultimate
from ultimate_mcp_server.constants import Provider, TaskType
from ultimate_mcp_server.exceptions import ProviderError, ToolError, ToolInputError
from ultimate_mcp_server.tools.base import (
with_cache,
with_error_handling,
with_retry,
with_tool_metrics,
)
from ultimate_mcp_server.tools.completion import generate_completion
from ultimate_mcp_server.utils import get_logger
logger = get_logger("ultimate_mcp_server.tools.ocr")
# Cache for storing preprocessed images and extracted text
OCR_CACHE = {}
# Check if required dependencies are available
def _check_ocr_dependencies():
"""Checks if OCR dependencies are available and returns a dictionary of requirements."""
requirements = {
"numpy": HAS_NUMPY,
"PIL": HAS_PIL,
"cv2": HAS_CV2,
"pytesseract": HAS_PYTESSERACT,
"pdf2image": HAS_PDF2IMAGE,
"pdfplumber": HAS_PDFPLUMBER,
"pymupdf": HAS_PYMUPDF
}
missing = [lib for lib, available in requirements.items() if not available]
if missing:
logger.warning(f"Some OCR dependencies are missing: {', '.join(missing)}")
logger.warning("OCR functionality may be limited. Install required packages with:")
packages = {
"numpy": "numpy",
"PIL": "pillow",
"cv2": "opencv-python-headless",
"pytesseract": "pytesseract",
"pdf2image": "pdf2image",
"pdfplumber": "pdfplumber",
"pymupdf": "pymupdf"
}
pip_command = f"pip install {' '.join(packages[lib] for lib in missing)}"
logger.warning(f" {pip_command}")
return requirements, missing
# Check dependencies early
OCR_REQUIREMENTS, MISSING_REQUIREMENTS = _check_ocr_dependencies()
# --- Helper functions for OCR processing ---
def _validate_file_path(file_path: str, expected_extension: Optional[str] = None) -> None:
"""
Validates a file path exists and optionally has the expected extension.
Args:
file_path: Path to the file to validate
expected_extension: Optional file extension to check (e.g., '.pdf')
Raises:
ToolInputError: If validation fails
"""
if not file_path:
raise ToolInputError("File path cannot be empty")
file_path = os.path.expanduser(os.path.normpath(file_path))
if not os.path.exists(file_path):
raise ToolInputError(f"File not found: {file_path}")
if not os.path.isfile(file_path):
raise ToolInputError(f"Path is not a file: {file_path}")
if expected_extension and not file_path.lower().endswith(expected_extension.lower()):
raise ToolInputError(f"File does not have the expected extension ({expected_extension}): {file_path}")
def _get_task_type_for_ocr(extraction_method: str = "hybrid") -> str:
"""
Returns the appropriate TaskType for OCR operations based on extraction method.
Args:
extraction_method: The extraction method being used
Returns:
The TaskType value as a string
"""
if extraction_method == "direct":
return TaskType.TEXT_EXTRACTION.value
elif extraction_method == "ocr":
return TaskType.OCR.value
else: # hybrid
return TaskType.OCR.value
def _handle_provider_error(e: Exception, operation: str) -> ToolError:
"""
Handles provider-specific errors and converts them to tool errors.
Args:
e: The exception that was raised
operation: Description of the operation that failed
Returns:
A ToolError with appropriate message
"""
if isinstance(e, ProviderError):
# Handle specific provider errors
return ToolError(f"Provider error during {operation}: {str(e)}")
else:
# Handle generic errors
return ToolError(f"Error during {operation}: {str(e)}")
def _preprocess_image(image: Image.Image, preprocessing_options: Optional[Dict[str, Any]] = None) -> Image.Image:
"""
Preprocesses an image for better OCR results.
Args:
image: PIL Image object
preprocessing_options: Dictionary of preprocessing options
- denoise: Whether to apply denoising (default: True)
- threshold: Thresholding method ('otsu', 'adaptive', 'none') (default: 'otsu')
- deskew: Whether to deskew the image (default: True)
- enhance_contrast: Whether to enhance contrast (default: True)
- enhance_brightness: Whether to enhance brightness (default: False)
- enhance_sharpness: Whether to enhance sharpness (default: False)
- apply_filters: List of filters to apply (default: [])
- resize_factor: Factor to resize the image by (default: 1.0)
Returns:
Preprocessed PIL Image object
"""
if not HAS_CV2 or not HAS_NUMPY or not HAS_PIL:
logger.warning("Image preprocessing requires opencv-python, numpy, and pillow. Using original image.")
return image
# Default preprocessing options
if preprocessing_options is None:
preprocessing_options = {
"denoise": True,
"threshold": "otsu",
"deskew": True,
"enhance_contrast": True,
"enhance_brightness": False,
"enhance_sharpness": False,
"apply_filters": [],
"resize_factor": 1.0
}
# Apply PIL enhancements before OpenCV processing if enabled
if HAS_PIL:
# Enhance brightness if requested
if preprocessing_options.get("enhance_brightness", False):
enhancer = ImageEnhance.Brightness(image)
# Increase brightness by 30%
image = enhancer.enhance(1.3)
# Enhance contrast if requested using PIL (in addition to OpenCV method)
if preprocessing_options.get("enhance_contrast", True):
enhancer = ImageEnhance.Contrast(image)
# Increase contrast by 40%
image = enhancer.enhance(1.4)
# Enhance sharpness if requested
if preprocessing_options.get("enhance_sharpness", False):
enhancer = ImageEnhance.Sharpness(image)
# Increase sharpness by 50%
image = enhancer.enhance(1.5)
# Apply filters if specified
filters = preprocessing_options.get("apply_filters", [])
for filter_name in filters:
if filter_name == "unsharp_mask":
image = image.filter(ImageFilter.UnsharpMask(radius=2, percent=150))
elif filter_name == "detail":
image = image.filter(ImageFilter.DETAIL)
elif filter_name == "edge_enhance":
image = image.filter(ImageFilter.EDGE_ENHANCE)
elif filter_name == "smooth":
image = image.filter(ImageFilter.SMOOTH)
# Convert PIL Image to OpenCV format
img = np.array(image)
if len(img.shape) == 3 and img.shape[2] == 3:
gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)
else:
gray = img
# Calculate optimal scaling based on image size and content
original_height, original_width = gray.shape[:2]
resize_factor = preprocessing_options.get("resize_factor", 1.0)
# Adaptive scaling based on image dimensions for optimal OCR
# For very small images, increase size; for very large images, reduce
if resize_factor == 1.0: # Only auto-adjust if user didn't specify
# Calculate the ideal size range for OCR (1500-3500 pixels on longest edge)
longest_edge = max(original_width, original_height)
if longest_edge < 1500:
# For small images, scale up to improve OCR
resize_factor = math.ceil(1500 / longest_edge * 10) / 10 # Round to nearest 0.1
elif longest_edge > 3500:
# For large images, scale down to improve performance
resize_factor = math.floor(3500 / longest_edge * 10) / 10 # Round to nearest 0.1
# Enhance contrast
if preprocessing_options.get("enhance_contrast", True):
gray = cv2.equalizeHist(gray)
# Apply thresholding
threshold_method = preprocessing_options.get("threshold", "otsu")
if threshold_method == "otsu":
_, img_thresholded = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU)
elif threshold_method == "adaptive":
# Calculate optimal block size based on image dimensions (odd number)
block_size = math.floor(min(gray.shape) / 30)
block_size = max(3, block_size)
if block_size % 2 == 0:
block_size += 1
img_thresholded = cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, block_size, 2)
else:
img_thresholded = gray
# Denoise
if preprocessing_options.get("denoise", True):
# Calculate optimal denoising parameters based on image size
h_param = math.ceil(10 * math.log10(min(original_width, original_height)))
img_denoised = cv2.fastNlMeansDenoising(img_thresholded, None, h_param, 7, 21)
else:
img_denoised = img_thresholded
# Deskew
if preprocessing_options.get("deskew", True) and HAS_NUMPY:
try:
coords = np.column_stack(np.where(img_denoised > 0))
angle = cv2.minAreaRect(coords)[-1]
if angle < -45:
angle = -(90 + angle)
else:
angle = -angle
# Rotate to correct skew if significant skew detected
if abs(angle) > 0.5:
(h, w) = img_denoised.shape[:2]
center = (w // 2, h // 2)
M = cv2.getRotationMatrix2D(center, angle, 1.0)
img_deskewed = cv2.warpAffine(img_denoised, M, (w, h), flags=cv2.INTER_CUBIC, borderMode=cv2.BORDER_REPLICATE)
else:
img_deskewed = img_denoised
except Exception as e:
logger.warning(f"Deskewing failed: {str(e)}. Using non-deskewed image.")
img_deskewed = img_denoised
else:
img_deskewed = img_denoised
# Resize if needed
if resize_factor != 1.0:
# Use ceiling to ensure we don't lose pixels in important small details
new_w = math.ceil(original_width * resize_factor)
new_h = math.ceil(original_height * resize_factor)
img_resized = cv2.resize(img_deskewed, (new_w, new_h), interpolation=cv2.INTER_CUBIC)
else:
img_resized = img_deskewed
# Convert back to PIL Image
return Image.fromarray(img_resized)
def _extract_text_with_ocr(image: Image.Image, ocr_language: str = "eng", ocr_config: str = "") -> str:
"""
Extracts text from an image using OCR.
Args:
image: PIL Image object
ocr_language: Language(s) for OCR (default: "eng")
ocr_config: Additional configuration for Tesseract
Returns:
Extracted text
"""
if not HAS_PYTESSERACT:
raise ToolError("pytesseract is required for OCR text extraction")
try:
custom_config = f"-l {ocr_language} {ocr_config}"
return pytesseract.image_to_string(image, config=custom_config)
except Exception as e:
logger.error(f"OCR extraction failed: {str(e)}")
raise ToolError(f"OCR extraction failed: {str(e)}") from e
def _extract_text_from_pdf_direct(file_path: str, start_page: int = 0, max_pages: int = 0) -> Tuple[List[str], bool]:
"""
Extracts text directly from a PDF file without OCR.
Args:
file_path: Path to the PDF file
start_page: First page to extract (0-indexed)
max_pages: Maximum number of pages to extract (0 = all)
Returns:
Tuple of (extracted_text_list, has_text)
"""
texts = []
has_text = False
if HAS_PDFPLUMBER:
try:
with pdfplumber.open(file_path) as pdf:
total_pages = len(pdf.pages)
end_page = total_pages if max_pages == 0 else min(start_page + max_pages, total_pages)
for i in range(start_page, end_page):
try:
page = pdf.pages[i]
text = page.extract_text(x_tolerance=3, y_tolerance=3)
if text and text.strip():
has_text = True
texts.append(text or "")
except Exception as e:
logger.warning(f"Error extracting text from page {i+1}: {str(e)}")
texts.append("")
except Exception as e:
logger.error(f"Error extracting text directly from PDF: {str(e)}")
raise ToolError(f"Failed to extract text directly from PDF: {str(e)}") from e
elif HAS_PYMUPDF:
try:
with pymupdf.open(file_path) as doc:
total_pages = len(doc)
end_page = total_pages if max_pages == 0 else min(start_page + max_pages, total_pages)
for i in range(start_page, end_page):
try:
page = doc[i]
text = page.get_text()
if text and text.strip():
has_text = True
texts.append(text or "")
except Exception as e:
logger.warning(f"Error extracting text from page {i+1}: {str(e)}")
texts.append("")
except Exception as e:
logger.error(f"Error extracting text directly from PDF: {str(e)}")
raise ToolError(f"Failed to extract text directly from PDF: {str(e)}") from e
else:
logger.warning("No PDF text extraction library available (pdfplumber or PyMuPDF)")
raise ToolError("No PDF text extraction library available. Install pdfplumber or PyMuPDF.")
return texts, has_text
def _convert_pdf_to_images(file_path, start_page=0, max_pages=0, dpi=300):
"""
Converts pages of a PDF file to PIL Image objects.
Args:
file_path: Path to the PDF file
start_page: First page to convert (0-indexed)
max_pages: Maximum number of pages to convert (0 = all)
dpi: DPI for rendering (default: 300)
Returns:
List of PIL Image objects
"""
if not HAS_PDF2IMAGE:
raise ToolError("pdf2image is required for PDF to image conversion")
try:
# Create a temporary directory to store intermediate images
# This helps with memory management for large PDFs
with tempfile.TemporaryDirectory() as temp_dir:
# pdf2image uses 1-based indexing
first_page = start_page + 1
last_page = None if max_pages == 0 else first_page + max_pages - 1
# Use the temp directory for output_folder
images = convert_from_path(
file_path,
dpi=dpi,
first_page=first_page,
last_page=last_page,
output_folder=temp_dir
)
return images
except Exception as e:
logger.error(f"PDF to image conversion failed: {str(e)}")
raise ToolError(f"Failed to convert PDF to images: {str(e)}") from e
def _convert_pdf_bytes_to_images(pdf_bytes, start_page=0, max_pages=0, dpi=300):
"""
Converts pages of a PDF from bytes to PIL Image objects.
Args:
pdf_bytes: PDF content as bytes
start_page: First page to convert (0-indexed)
max_pages: Maximum number of pages to convert (0 = all)
dpi: DPI for rendering (default: 300)
Returns:
List of PIL Image objects
"""
if not HAS_PDF2IMAGE:
raise ToolError("pdf2image is required for PDF to image conversion")
try:
# Create a temporary directory to store intermediate images
# This helps with memory management for large PDFs
with tempfile.TemporaryDirectory() as temp_dir:
# pdf2image uses 1-based indexing
first_page = start_page + 1
last_page = None if max_pages == 0 else first_page + max_pages - 1
# Use the temp directory for output_folder
images = convert_from_bytes(
pdf_bytes,
dpi=dpi,
first_page=first_page,
last_page=last_page,
output_folder=temp_dir
)
return images
except Exception as e:
logger.error(f"PDF bytes to image conversion failed: {str(e)}")
raise ToolError(f"Failed to convert PDF bytes to images: {str(e)}") from e
def _generate_cache_key(data, prefix="ocr"):
"""Generate a cache key for the given data."""
if isinstance(data, str) and os.path.exists(data):
# For file paths, use mtime and size
stat = os.stat(data)
key_data = f"{data}:{stat.st_mtime}:{stat.st_size}"
elif isinstance(data, Image.Image):
# For PIL images, convert to bytes and hash
img_bytes = io.BytesIO()
data.save(img_bytes, format=data.format or 'PNG')
key_data = img_bytes.getvalue()
elif isinstance(data, dict):
# For dictionaries, convert to JSON
key_data = json.dumps(data, sort_keys=True)
else:
# For other data, use string representation
key_data = str(data)
# Generate hash
h = hashlib.md5(key_data.encode() if isinstance(key_data, str) else key_data)
# Add a UUID component for uniqueness across process restarts
unique_id = str(uuid.uuid4())[:8]
return f"{prefix}_{h.hexdigest()}_{unique_id}"
def _split_text_into_chunks(text, max_chunk_size=8000, overlap=200):
"""
Splits text into chunks of specified maximum size with overlap.
Args:
text: Text to split
max_chunk_size: Maximum chunk size in characters
overlap: Overlap between chunks in characters
Returns:
List of text chunks
"""
if not text:
return []
# Ensure reasonable values
max_chunk_size = max(1000, min(max_chunk_size, 15000))
overlap = max(50, min(overlap, max_chunk_size // 4))
# Split by paragraphs first
paragraphs = re.split(r'\n\s*\n', text)
chunks = []
current_chunk = []
current_length = 0
for paragraph in paragraphs:
para_length = len(paragraph)
if current_length + para_length <= max_chunk_size:
# Paragraph fits in current chunk
current_chunk.append(paragraph)
current_length += para_length + 2 # +2 for the newlines
else:
# Paragraph doesn't fit
if current_chunk:
# Save current chunk
chunks.append("\n\n".join(current_chunk))
if para_length <= max_chunk_size:
# Start new chunk with this paragraph
current_chunk = [paragraph]
current_length = para_length + 2
else:
# Paragraph too large, split into sentences
sentences = re.split(r'(?<=[.!?])\s+', paragraph)
current_chunk = []
current_length = 0
for sentence in sentences:
sentence_length = len(sentence)
if current_length + sentence_length <= max_chunk_size:
# Sentence fits in current chunk
current_chunk.append(sentence)
current_length += sentence_length + 1 # +1 for the space
else:
# Sentence doesn't fit
if current_chunk:
# Save current chunk
chunks.append(" ".join(current_chunk))
if sentence_length <= max_chunk_size:
# Start new chunk with this sentence
current_chunk = [sentence]
current_length = sentence_length + 1
else:
# Sentence too large, split by words
words = sentence.split()
current_chunk = []
current_length = 0
current_part = []
part_length = 0
for word in words:
word_length = len(word)
if part_length + word_length + 1 <= max_chunk_size:
current_part.append(word)
part_length += word_length + 1 # +1 for the space
else:
if current_part:
chunks.append(" ".join(current_part))
current_part = [word]
part_length = word_length + 1
if current_part:
current_chunk = current_part
current_length = part_length
# Add the last chunk if it exists
if current_chunk:
chunks.append("\n\n".join(current_chunk) if len(current_chunk) > 1 else current_chunk[0])
# Add overlap between chunks
result = []
prev_end = ""
for i, chunk in enumerate(chunks):
if i > 0 and prev_end:
# Find a good overlap point (try to break at paragraph or sentence)
overlap_text = prev_end
if "\n\n" in overlap_text:
parts = overlap_text.split("\n\n")
if len(parts) > 1:
overlap_text = parts[-1]
# Prepend overlap to current chunk
chunk = overlap_text + " " + chunk
# Save end of current chunk for next iteration
prev_end = chunk[-overlap:] if len(chunk) > overlap else chunk
result.append(chunk)
return result
def _detect_tables(image: Image.Image) -> List[Tuple[int, int, int, int]]:
"""
Detects potential tables in an image.
Args:
image: PIL Image object
Returns:
List of detected table regions as (x, y, width, height) tuples
"""
if not HAS_CV2 or not HAS_NUMPY:
return []
# Convert PIL Image to OpenCV format
img = np.array(image)
if len(img.shape) == 3 and img.shape[2] == 3:
gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)
else:
gray = img
# Apply thresholding and morphological operations
_, thresh = cv2.threshold(gray, 150, 255, cv2.THRESH_BINARY_INV)
# Create a kernel for dilation
kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (5, 5))
dilated = cv2.dilate(thresh, kernel, iterations=5)
# Find contours
contours, _ = cv2.findContours(dilated, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
# Filter contours to find potential tables
table_regions = []
for contour in contours:
x, y, w, h = cv2.boundingRect(contour)
# Tables usually have a certain aspect ratio and size
aspect_ratio = w / h
area = w * h
img_area = img.shape[0] * img.shape[1]
if 0.5 <= aspect_ratio <= 3.0 and area > img_area * 0.05:
table_regions.append((x, y, w, h))
return table_regions
def _crop_image(image: Image.Image, region: Tuple[int, int, int, int]) -> Image.Image:
"""
Crops an image to the specified region.
Args:
image: PIL Image object
region: Tuple of (x, y, width, height)
Returns:
Cropped PIL Image object
"""
x, y, width, height = region
return image.crop((x, y, x + width, y + height))
def _is_text_mostly_noise(text, noise_threshold=0.3):
"""Determine if extracted text is mostly noise based on character distribution."""
if not text or len(text) < 10:
return False
# Calculate the ratio of non-alphanumeric and non-punctuation characters
total_chars = len(text)
valid_chars = sum(1 for c in text if c.isalnum() or c.isspace() or c in '.,;:!?"-\'()[]{}')
noise_ratio = 1 - (valid_chars / total_chars)
return noise_ratio > noise_threshold
def _is_likely_header_or_footer(text, line_length_threshold=50):
"""Determine if a text line is likely a header or footer."""
text = text.strip()
if len(text) == 0:
return False
# Short lines with page numbers
if len(text) < line_length_threshold and re.search(r'\b\d+\b', text):
return True
# Common header/footer patterns
patterns = [
r'^\d+$', # Just a page number
r'^Page\s+\d+(\s+of\s+\d+)?$', # Page X of Y
r'^[\w\s]+\s+\|\s+\d+$', # Title | Page
r'^\w+\s+\d{1,2},?\s+\d{4}$', # Date format
r'^Copyright', # Copyright notices
r'^\w+\s+\d{1,2}(st|nd|rd|th)?,?\s+\d{4}$', # Date with ordinal
r'^\d{1,2}/\d{1,2}/\d{2,4}$' # Date in MM/DD/YY format
]
for pattern in patterns:
if re.search(pattern, text, re.IGNORECASE):
return True
return False
def _remove_headers_and_footers(text, max_line_length=70):
"""
Removes headers and footers from text.
Args:
text: Text to process
max_line_length: Maximum length for a line to be considered a header/footer
Returns:
Text with headers and footers removed
"""
if not text:
return text
# Split text into lines
lines = text.splitlines()
result = []
for _i, line in enumerate(lines):
# Skip empty lines
if not line.strip():
result.append(line)
continue
# Check if line is likely a header or footer
if len(line.strip()) <= max_line_length and _is_likely_header_or_footer(line):
# Replace with empty line to maintain spacing
result.append("")
continue
result.append(line)
# Join lines back together
return "\n".join(result)
async def _process_text_chunk(chunk: str, reformat_as_markdown: bool = False, remove_headers: bool = False) -> str:
"""
Processes a chunk of OCR text with LLM enhancement.
Args:
chunk: Text chunk to process
reformat_as_markdown: Whether to format as markdown
remove_headers: Whether to remove headers and footers
Returns:
Enhanced text chunk
"""
if not chunk.strip():
return ""
# First apply simple rule-based fixes
cleaned_text = chunk
# Fix hyphenated words at line breaks
cleaned_text = re.sub(r'(\w+)-\s*\n\s*(\w+)', lambda m: f"{m.group(1)}{m.group(2)}", cleaned_text)
# Remove obvious noise
if _is_text_mostly_noise(cleaned_text):
logger.warning("Text chunk appears to be mostly noise, applying aggressive cleaning")
# Replace unusual characters with spaces
cleaned_text = re.sub(r'[^\w\s.,;:!?"\'\(\)\[\]\{\}-]', ' ', cleaned_text)
# Normalize spaces
cleaned_text = re.sub(r'\s+', ' ', cleaned_text)
# Remove headers and footers if requested
if remove_headers:
cleaned_text = _remove_headers_and_footers(cleaned_text)
# Prepare LLM enhancement prompt
if reformat_as_markdown:
prompt = f"""Correct OCR errors in this text and format it as markdown. Follow these instructions:
1. Fix OCR-induced errors:
- Correct words split across line breaks (e.g., "cor- rect" → "correct")
- Fix typos like 'rn' misread as 'm', '0' misread as 'O', etc.
- Merge split paragraphs but preserve intentional paragraph breaks
- Use context and common sense to correct errors
2. Format as markdown:
- Convert headings to markdown headings (# for main title, ## for subtitles, etc.)
- Format lists as proper markdown lists
- Use emphasis (*italic*) and strong (**bold**) where appropriate
- Create tables using markdown syntax if tabular data is detected
- For code or equations, use appropriate markdown formatting
3. Clean up formatting:
- Remove unnecessary line breaks within paragraphs
- Preserve paragraph structure
- Remove duplicated text
- {"Remove headers, footers, and page numbers" if remove_headers else "Preserve all content including headers/footers"}
4. Preserve the original content's meaning and information.
Here is the text to correct and format:
```
{cleaned_text}
```
Provide ONLY the corrected markdown text with no explanations or comments.
"""
else:
prompt = f"""Correct OCR errors in this text. Follow these instructions:
1. Fix OCR-induced errors:
- Correct words split across line breaks (e.g., "cor- rect" → "correct")
- Fix typos like 'rn' misread as 'm', '0' misread as 'O', etc.
- Merge split paragraphs but preserve intentional paragraph breaks
- Use context and common sense to correct errors
2. Clean up formatting:
- Remove unnecessary line breaks within paragraphs
- Preserve paragraph structure
- Remove duplicated text
- {"Remove headers, footers, and page numbers" if remove_headers else "Preserve all content including headers/footers"}
3. Preserve the original content's meaning and information.
Here is the text to correct:
```
{cleaned_text}
```
Provide ONLY the corrected text with no explanations or comments.
"""
try:
# Use generate_completion to process the text
task_type = TaskType.TEXT_ENHANCEMENT.value
result = await generate_completion(
prompt=prompt,
provider=Provider.ANTHROPIC.value, # Default to Anthropic for high-quality text processing
temperature=0.2, # Low temperature for consistent results
max_tokens=len(cleaned_text) + 1000, # Allow some expansion for formatting
task_type=task_type
)
if not result or not result.get("text"):
logger.warning("LLM text enhancement returned empty result")
return cleaned_text
enhanced_text = result["text"]
# Remove any "Here is the corrected..." prefixes that LLMs sometimes add
enhanced_text = re.sub(r'^(Here is|The corrected|Here\'s)[^:]*:?\s*', '', enhanced_text, flags=re.IGNORECASE)
return enhanced_text
except ProviderError as e:
logger.error(f"Provider error during text enhancement: {str(e)}")
# Fall back to the cleaned text
return cleaned_text
except Exception as e:
logger.error(f"Error during LLM text enhancement: {str(e)}")
# Fall back to the cleaned text
return cleaned_text
# --- Main OCR tool functions ---
@with_cache(ttl=24 * 60 * 60) # Cache for 24 hours
@with_tool_metrics
@with_retry(max_retries=3, retry_delay=1)
@with_error_handling
async def extract_text_from_pdf(
file_path: str,
extraction_method: str = "hybrid",
max_pages: int = 0,
skip_pages: int = 0,
preprocessing_options: Optional[Dict[str, Any]] = None,
ocr_language: str = "eng",
reformat_as_markdown: bool = False,
suppress_headers: bool = False,
assess_quality: bool = False,
dpi: int = 300
) -> Dict[str, Any]:
"""
Extracts and enhances text from a PDF document.
This tool can use multiple extraction methods: direct text extraction from the PDF,
OCR-based extraction, or a hybrid approach that uses direct extraction when possible
and falls back to OCR when necessary. The extracted text is then enhanced using an
LLM to correct OCR errors and optionally format the output as markdown.
Args:
file_path: Path to the PDF file
extraction_method: Method to use for text extraction:
- "direct": Extract text directly from the PDF (fastest, but may fail for scanned PDFs)
- "ocr": Always use OCR (slower but works for scanned PDFs)
- "hybrid": Try direct extraction first, fall back to OCR if needed (default)
max_pages: Maximum number of pages to process (0 = all pages)
skip_pages: Number of pages to skip from the beginning (0-indexed)
preprocessing_options: Dictionary of options for image preprocessing:
- denoise: Whether to apply denoising (default: True)
- threshold: Thresholding method ('otsu', 'adaptive', 'none') (default: 'otsu')
- deskew: Whether to deskew the image (default: True)
- enhance_contrast: Whether to enhance contrast (default: True)
- resize_factor: Factor to resize the image (default: 1.0)
ocr_language: Language(s) for OCR, e.g., "eng" or "eng+fra" (default: "eng")
reformat_as_markdown: Whether to format the output as markdown (default: False)
suppress_headers: Whether to remove headers, footers, and page numbers (default: False)
assess_quality: Whether to assess the quality of the OCR improvement (default: False)
dpi: DPI for PDF rendering when using OCR (default: 300)
Returns:
A dictionary containing:
{
"success": true,
"text": "The extracted and enhanced text...",
"raw_text": "The original OCR text before enhancement...",
"pages_processed": 5,
"extraction_method_used": "hybrid",
"file_path": "/path/to/document.pdf",
"quality_metrics": { # Only if assess_quality=True
"score": 85,
"explanation": "Explanation of quality score..."
},
"processing_time": 12.34 # Seconds
}
Raises:
ToolInputError: If the file path is invalid or the file is not a PDF
ToolError: If text extraction fails
"""
start_time = time.time()
# Validate file path
_validate_file_path(file_path, expected_extension=".pdf")
# Check extraction method
valid_methods = ["direct", "ocr", "hybrid"]
if extraction_method not in valid_methods:
raise ToolInputError(
f"Invalid extraction method: '{extraction_method}'. Must be one of: {', '.join(valid_methods)}"
)
# Check dependencies based on extraction method
if extraction_method in ["ocr", "hybrid"]:
if not HAS_PDF2IMAGE or not HAS_PYTESSERACT:
logger.warning(f"OCR extraction requires pdf2image and pytesseract. {extraction_method} may fail.")
if extraction_method in ["direct", "hybrid"]:
if not HAS_PDFPLUMBER and not HAS_PYMUPDF:
logger.warning("Direct extraction requires pdfplumber or PyMuPDF.")
# Initialize result
result = {
"success": False,
"file_path": file_path,
"pages_processed": 0,
"extraction_method_used": extraction_method
}
method_used = extraction_method
raw_text_list = []
extracted_text_list = []
has_direct_text = False
try:
# Step 1: Extract text
if extraction_method in ["direct", "hybrid"]:
try:
logger.info(f"Attempting direct text extraction from PDF: {file_path}")
direct_text_list, has_direct_text = _extract_text_from_pdf_direct(
file_path,
start_page=skip_pages,
max_pages=max_pages
)
raw_text_list = direct_text_list
logger.info(f"Direct text extraction {'succeeded' if has_direct_text else 'failed'}")
if has_direct_text and extraction_method == "direct":
# If direct extraction found text and that's the requested method, we're done
method_used = "direct"
extracted_text_list = direct_text_list
logger.info(f"Using direct extraction result with {len(extracted_text_list)} pages")
elif has_direct_text and extraction_method == "hybrid":
# If hybrid mode and direct extraction worked, use it
method_used = "direct"
extracted_text_list = direct_text_list
logger.info(f"Using direct extraction result with {len(extracted_text_list)} pages (hybrid mode)")
elif extraction_method == "direct" and not has_direct_text:
# If direct mode but no text found, we fail
raise ToolError("Direct text extraction failed to find text in the PDF")
# If hybrid mode and no text found, fall back to OCR
if extraction_method == "hybrid" and not has_direct_text:
logger.info("No text found via direct extraction, falling back to OCR (hybrid mode)")
method_used = "ocr"
# Continue to OCR extraction below
except Exception as e:
logger.error(f"Direct text extraction failed: {str(e)}")
if extraction_method == "direct":
raise ToolError(f"Direct text extraction failed: {str(e)}") from e
logger.info("Falling back to OCR extraction")
method_used = "ocr"
# Step 2: OCR extraction if needed
if method_used == "ocr" or extraction_method == "ocr":
method_used = "ocr"
logger.info(f"Performing OCR-based text extraction on PDF: {file_path}")
# Convert PDF to images
images = _convert_pdf_to_images(
file_path,
start_page=skip_pages,
max_pages=max_pages,
dpi=dpi
)
# Extract text using OCR
raw_text_list = []
with ThreadPoolExecutor() as executor:
# Preprocess images in parallel
preprocessed_images = list(executor.map(
lambda img: _preprocess_image(img, preprocessing_options),
images
))
# Extract text in parallel
ocr_config = ""
ocr_results = list(executor.map(
lambda img: _extract_text_with_ocr(img, ocr_language, ocr_config),
preprocessed_images
))
extracted_text_list = ocr_results
raw_text_list = ocr_results
logger.info(f"OCR extraction completed for {len(extracted_text_list)} pages")
# Step 3: Process extracted text
logger.info("Processing extracted text with LLM enhancement")
# Combine text from pages
full_raw_text = "\n\n".join(raw_text_list)
# Split into chunks for LLM processing
chunks = _split_text_into_chunks(full_raw_text)
logger.info(f"Text split into {len(chunks)} chunks for LLM processing")
# Process chunks in parallel
enhanced_chunks = await asyncio.gather(*[
_process_text_chunk(chunk, reformat_as_markdown, suppress_headers)
for chunk in chunks
])
# Combine chunks
enhanced_text = "\n\n".join(enhanced_chunks)
# Step 4: Assess quality if requested
quality_metrics = None
if assess_quality:
logger.info("Assessing quality of text enhancement")
quality_metrics = await _assess_text_quality(full_raw_text, enhanced_text)
# Prepare final result
processing_time = time.time() - start_time
result.update({
"success": True,
"text": enhanced_text,
"raw_text": full_raw_text,
"pages_processed": len(raw_text_list),
"extraction_method_used": method_used,
"processing_time": processing_time
})
if quality_metrics:
result["quality_metrics"] = quality_metrics
logger.info(f"Text extraction and enhancement completed successfully in {processing_time:.2f}s")
return result
except Exception as e:
logger.error(f"Error in extract_text_from_pdf: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to extract and enhance text from PDF: {str(e)}") from e
@with_cache(ttl=24 * 60 * 60) # Cache for 24 hours
@with_tool_metrics
@with_retry(max_retries=3, retry_delay=1)
@with_error_handling
async def extract_text_from_pdf_bytes(
pdf_bytes: bytes,
extraction_method: str = "hybrid",
max_pages: int = 0,
skip_pages: int = 0,
preprocessing_options: Optional[Dict[str, Any]] = None,
ocr_language: str = "eng",
reformat_as_markdown: bool = False,
suppress_headers: bool = False,
assess_quality: bool = False,
dpi: int = 300
) -> Dict[str, Any]:
"""
Extracts and enhances text from PDF bytes data.
This tool works like extract_text_from_pdf but accepts PDF data as bytes instead of a file path.
It can use multiple extraction methods and enhance the extracted text using an LLM.
Args:
pdf_bytes: PDF content as bytes
extraction_method: Method to use for text extraction:
- "direct": Extract text directly from the PDF (fastest, but may fail for scanned PDFs)
- "ocr": Always use OCR (slower but works for scanned PDFs)
- "hybrid": Try direct extraction first, fall back to OCR if needed (default)
max_pages: Maximum number of pages to process (0 = all pages)
skip_pages: Number of pages to skip from the beginning (0-indexed)
preprocessing_options: Dictionary of options for image preprocessing
ocr_language: Language(s) for OCR, e.g., "eng" or "eng+fra" (default: "eng")
reformat_as_markdown: Whether to format the output as markdown (default: False)
suppress_headers: Whether to remove headers, footers, and page numbers (default: False)
assess_quality: Whether to assess the quality of the OCR improvement (default: False)
dpi: DPI for PDF rendering when using OCR (default: 300)
Returns:
A dictionary with the extracted and enhanced text, same format as extract_text_from_pdf
Raises:
ToolInputError: If the PDF bytes are invalid
ToolError: If text extraction fails
"""
start_time = time.time()
# Validate input
if not pdf_bytes:
raise ToolInputError("PDF bytes cannot be empty")
# Check extraction method
valid_methods = ["direct", "ocr", "hybrid"]
if extraction_method not in valid_methods:
raise ToolInputError(
f"Invalid extraction method: '{extraction_method}'. Must be one of: {', '.join(valid_methods)}"
)
# Check dependencies based on extraction method
if extraction_method in ["ocr", "hybrid"]:
if not HAS_PDF2IMAGE or not HAS_PYTESSERACT:
logger.warning(f"OCR extraction requires pdf2image and pytesseract. {extraction_method} may fail.")
if extraction_method in ["direct", "hybrid"]:
if not HAS_PDFPLUMBER and not HAS_PYMUPDF:
logger.warning("Direct extraction requires pdfplumber or PyMuPDF.")
# Initialize result
result = {
"success": False,
"pages_processed": 0,
"extraction_method_used": extraction_method
}
method_used = extraction_method
raw_text_list = []
extracted_text_list = []
has_direct_text = False
try:
# Create a temporary file for processing
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as temp_pdf:
temp_path = temp_pdf.name
temp_pdf.write(pdf_bytes)
temp_pdf.flush()
try:
# Step 1: Extract text
if extraction_method in ["direct", "hybrid"]:
try:
logger.info("Attempting direct text extraction from PDF bytes")
direct_text_list, has_direct_text = _extract_text_from_pdf_direct(
temp_path,
start_page=skip_pages,
max_pages=max_pages
)
raw_text_list = direct_text_list
logger.info(f"Direct text extraction {'succeeded' if has_direct_text else 'failed'}")
if has_direct_text and extraction_method == "direct":
method_used = "direct"
extracted_text_list = direct_text_list
logger.info(f"Using direct extraction result with {len(extracted_text_list)} pages")
elif has_direct_text and extraction_method == "hybrid":
method_used = "direct"
extracted_text_list = direct_text_list
logger.info(f"Using direct extraction result with {len(extracted_text_list)} pages (hybrid mode)")
elif extraction_method == "direct" and not has_direct_text:
raise ToolError("Direct text extraction failed to find text in the PDF")
if extraction_method == "hybrid" and not has_direct_text:
logger.info("No text found via direct extraction, falling back to OCR (hybrid mode)")
method_used = "ocr"
except Exception as e:
logger.error(f"Direct text extraction failed: {str(e)}")
if extraction_method == "direct":
raise ToolError(f"Direct text extraction failed: {str(e)}") from e
logger.info("Falling back to OCR extraction")
method_used = "ocr"
# Step 2: OCR extraction if needed
if method_used == "ocr" or extraction_method == "ocr":
method_used = "ocr"
logger.info("Performing OCR-based text extraction on PDF bytes")
# Convert PDF bytes to images
images = _convert_pdf_bytes_to_images(
pdf_bytes,
start_page=skip_pages,
max_pages=max_pages,
dpi=dpi
)
# Extract text using OCR
raw_text_list = []
with ThreadPoolExecutor() as executor:
# Preprocess images in parallel
preprocessed_images = list(executor.map(
lambda img: _preprocess_image(img, preprocessing_options),
images
))
# Extract text in parallel
ocr_config = ""
ocr_results = list(executor.map(
lambda img: _extract_text_with_ocr(img, ocr_language, ocr_config),
preprocessed_images
))
extracted_text_list = ocr_results
raw_text_list = ocr_results
logger.info(f"OCR extraction completed for {len(extracted_text_list)} pages")
# Step 3: Process extracted text
logger.info("Processing extracted text with LLM enhancement")
# Combine text from pages
full_raw_text = "\n\n".join(raw_text_list)
# Split into chunks for LLM processing
chunks = _split_text_into_chunks(full_raw_text)
logger.info(f"Text split into {len(chunks)} chunks for LLM processing")
# Process chunks in parallel
enhanced_chunks = await asyncio.gather(*[
_process_text_chunk(chunk, reformat_as_markdown, suppress_headers)
for chunk in chunks
])
# Combine chunks
enhanced_text = "\n\n".join(enhanced_chunks)
# Step 4: Assess quality if requested
quality_metrics = None
if assess_quality:
logger.info("Assessing quality of text enhancement")
quality_metrics = await _assess_text_quality(full_raw_text, enhanced_text)
# Prepare final result
processing_time = time.time() - start_time
result.update({
"success": True,
"text": enhanced_text,
"raw_text": full_raw_text,
"pages_processed": len(raw_text_list),
"extraction_method_used": method_used,
"processing_time": processing_time
})
if quality_metrics:
result["quality_metrics"] = quality_metrics
logger.info(f"Text extraction and enhancement completed successfully in {processing_time:.2f}s")
return result
finally:
# Clean up temporary file
try:
os.unlink(temp_path)
except Exception as e:
logger.warning(f"Failed to remove temporary file: {str(e)}")
except Exception as e:
logger.error(f"Error in extract_text_from_pdf_bytes: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to extract and enhance text from PDF bytes: {str(e)}") from e
@with_cache(ttl=24 * 60 * 60) # Cache for 24 hours
@with_tool_metrics
@with_retry(max_retries=2, retry_delay=1)
@with_error_handling
async def process_image_ocr(
image_path: Optional[str] = None,
image_data: Optional[str] = None,
preprocessing_options: Optional[Dict[str, Any]] = None,
ocr_language: str = "eng",
reformat_as_markdown: bool = False,
assess_quality: bool = False
) -> Dict[str, Any]:
"""
Processes an image with OCR and enhances the extracted text.
This tool accepts either a path to an image file or base64-encoded image data,
performs OCR on the image, and then enhances the extracted text using an LLM.
Args:
image_path: Path to the image file (mutually exclusive with image_data)
image_data: Base64-encoded image data (mutually exclusive with image_path)
preprocessing_options: Dictionary of options for image preprocessing:
- denoise: Whether to apply denoising (default: True)
- threshold: Thresholding method ('otsu', 'adaptive', 'none') (default: 'otsu')
- deskew: Whether to deskew the image (default: True)
- enhance_contrast: Whether to enhance contrast (default: True)
- resize_factor: Factor to resize the image (default: 1.0)
ocr_language: Language(s) for OCR, e.g., "eng" or "eng+fra" (default: "eng")
reformat_as_markdown: Whether to format the output as markdown (default: False)
assess_quality: Whether to assess the quality of the OCR improvement (default: False)
Returns:
A dictionary containing:
{
"success": true,
"text": "The extracted and enhanced text...",
"raw_text": "The original OCR text before enhancement...",
"table_detected": false, # Whether a table was detected in the image
"quality_metrics": { # Only if assess_quality=True
"score": 85,
"explanation": "Explanation of quality score..."
},
"processing_time": 3.45 # Seconds
}
Raises:
ToolInputError: If input is invalid
ToolError: If processing fails
"""
start_time = time.time()
# Check dependencies
if not HAS_PIL or not HAS_PYTESSERACT:
missing = []
if not HAS_PIL:
missing.append("pillow")
if not HAS_PYTESSERACT:
missing.append("pytesseract")
raise ToolError(f"Required dependencies missing: {', '.join(missing)}")
# Validate input
if not image_path and not image_data:
raise ToolInputError("Either image_path or image_data must be provided")
if image_path and image_data:
raise ToolInputError("Only one of image_path or image_data should be provided")
try:
# Load image
if image_path:
_validate_file_path(image_path)
image = Image.open(image_path)
else:
# Decode base64 image data
try:
image_bytes = base64.b64decode(image_data)
image = Image.open(io.BytesIO(image_bytes))
except Exception as e:
raise ToolInputError(f"Invalid base64 image data: {str(e)}") from e
# Preprocess image
logger.info("Preprocessing image for OCR")
preprocessed_image = _preprocess_image(image, preprocessing_options)
# Detect tables
table_regions = _detect_tables(preprocessed_image)
table_detected = len(table_regions) > 0
logger.info(f"Table detection: {len(table_regions)} potential tables found")
# Extract text with OCR
logger.info(f"Performing OCR with language(s): {ocr_language}")
raw_text = _extract_text_with_ocr(preprocessed_image, ocr_language)
# Process tables separately if detected
table_texts = []
if table_detected and HAS_CV2:
logger.info("Processing detected tables separately")
for i, region in enumerate(table_regions):
try:
table_image = _crop_image(preprocessed_image, region)
# Use a different preprocessing for tables (less aggressive)
table_options = {"denoise": True, "threshold": "adaptive", "deskew": False}
processed_table_image = _preprocess_image(table_image, table_options)
table_text = _extract_text_with_ocr(processed_table_image, ocr_language)
if table_text.strip():
table_texts.append(f"\n\nTable {i+1}:\n{table_text}\n")
except Exception as e:
logger.warning(f"Error processing table {i+1}: {str(e)}")
# Include table texts with the main text
if table_texts:
raw_text += "\n\n" + "\n".join(table_texts)
# Process with LLM
logger.info("Processing extracted text with LLM enhancement")
enhanced_text = await _process_text_chunk(raw_text, reformat_as_markdown, suppress_headers=False)
# Assess quality if requested
quality_metrics = None
if assess_quality:
logger.info("Assessing quality of text enhancement")
quality_metrics = await _assess_text_quality(raw_text, enhanced_text)
# Prepare result
processing_time = time.time() - start_time
result = {
"success": True,
"text": enhanced_text,
"raw_text": raw_text,
"table_detected": table_detected,
"processing_time": processing_time
}
if quality_metrics:
result["quality_metrics"] = quality_metrics
logger.info(f"Image OCR processing completed in {processing_time:.2f}s")
return result
except Exception as e:
logger.error(f"Error in process_image_ocr: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to process image with OCR: {str(e)}") from e
@with_cache(ttl=24 * 60 * 60) # Cache for 24 hours
@with_tool_metrics
@with_retry(max_retries=2, retry_delay=1)
@with_error_handling
async def enhance_ocr_text(
ocr_text: str,
reformat_as_markdown: bool = False,
remove_headers: bool = False,
detect_tables: bool = True,
assess_quality: bool = False
) -> Dict[str, Any]:
"""
Enhances existing OCR text using an LLM to correct errors and improve formatting.
This tool takes OCR text (e.g., from a different OCR engine) and uses an LLM to
correct errors, improve formatting, and optionally convert to markdown.
Args:
ocr_text: The OCR text to enhance
reformat_as_markdown: Whether to format the output as markdown (default: False)
remove_headers: Whether to remove headers, footers, and page numbers (default: False)
detect_tables: Whether to attempt to detect and format tables (default: True)
assess_quality: Whether to assess the quality of the OCR improvement (default: False)
Returns:
A dictionary containing:
{
"success": true,
"text": "The enhanced text...",
"raw_text": "The original OCR text...",
"quality_metrics": { # Only if assess_quality=True
"score": 85,
"explanation": "Explanation of quality score..."
},
"processing_time": 2.34 # Seconds
}
Raises:
ToolInputError: If the OCR text is empty
ToolError: If enhancement fails
"""
start_time = time.time()
# Validate input
if not ocr_text or not isinstance(ocr_text, str):
raise ToolInputError("OCR text must be a non-empty string")
try:
# Split into chunks if large
if len(ocr_text) > 10000:
logger.info(f"Splitting large OCR text ({len(ocr_text)} chars) into chunks")
chunks = _split_text_into_chunks(ocr_text)
# Process chunks in parallel
enhanced_chunks = await asyncio.gather(*[
_process_text_chunk(chunk, reformat_as_markdown, remove_headers)
for chunk in chunks
])
# Combine chunks
enhanced_text = "\n\n".join(enhanced_chunks)
logger.info(f"Processed {len(chunks)} text chunks")
else:
# Process directly if small enough
enhanced_text = await _process_text_chunk(ocr_text, reformat_as_markdown, remove_headers)
# Detect and format tables if requested
if detect_tables and reformat_as_markdown:
logger.info("Attempting table detection and formatting")
enhanced_text = await _format_tables_in_text(enhanced_text)
# Assess quality if requested
quality_metrics = None
if assess_quality:
logger.info("Assessing quality of text enhancement")
quality_metrics = await _assess_text_quality(ocr_text, enhanced_text)
# Prepare result
processing_time = time.time() - start_time
result = {
"success": True,
"text": enhanced_text,
"raw_text": ocr_text,
"processing_time": processing_time
}
if quality_metrics:
result["quality_metrics"] = quality_metrics
logger.info(f"OCR text enhancement completed in {processing_time:.2f}s")
return result
except Exception as e:
logger.error(f"Error in enhance_ocr_text: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to enhance OCR text: {str(e)}") from e
@with_tool_metrics
@with_retry(max_retries=2, retry_delay=1.0)
@with_error_handling
async def analyze_pdf_structure(
file_path: str,
extract_metadata: bool = True,
extract_outline: bool = True,
extract_fonts: bool = False,
extract_images: bool = False,
estimate_ocr_needs: bool = True
) -> Dict[str, Any]:
"""
Analyzes the structure of a PDF file without performing full text extraction.
This tool examines a PDF file and provides information about its structure,
including metadata, outline (table of contents), fonts, embedded images,
and an assessment of whether OCR would be beneficial.
Args:
file_path: Path to the PDF file
extract_metadata: Whether to extract document metadata (default: True)
extract_outline: Whether to extract the document outline/TOC (default: True)
extract_fonts: Whether to extract font information (default: False)
extract_images: Whether to extract information about embedded images (default: False)
estimate_ocr_needs: Whether to estimate if OCR would benefit this PDF (default: True)
Returns:
A dictionary containing:
{
"success": true,
"file_path": "/path/to/document.pdf",
"page_count": 42,
"metadata": { # Only if extract_metadata=True
"title": "Document Title",
"author": "Author Name",
"subject": "Document Subject",
"keywords": "keyword1, keyword2",
"creator": "Creator Application",
"producer": "Producer Application",
"creation_date": "2023-01-01T12:00:00",
"modification_date": "2023-02-01T13:00:00"
},
"outline": [ # Only if extract_outline=True
{
"title": "Chapter 1",
"page": 5,
"children": [
{"title": "Section 1.1", "page": 6, "children": []}
]
},
{"title": "Chapter 2", "page": 15, "children": []}
],
"font_info": { # Only if extract_fonts=True
"total_fonts": 3,
"embedded_fonts": 2,
"font_names": ["Arial", "Times New Roman", "Courier"]
},
"image_info": { # Only if extract_images=True
"total_images": 12,
"image_types": {"jpeg": 8, "png": 4},
"average_size": "120kb"
},
"ocr_assessment": { # Only if estimate_ocr_needs=True
"needs_ocr": false,
"confidence": "high",
"reason": "PDF contains extractable text throughout"
},
"processing_time": 1.23 # Seconds
}
Raises:
ToolInputError: If the file path is invalid or the file is not a PDF
ToolError: If analysis fails
"""
start_time = time.time()
# Validate file path
_validate_file_path(file_path, expected_extension=".pdf")
# Check for required libraries
pdf_lib_available = False
if HAS_PYMUPDF:
pdf_lib = "pymupdf"
pdf_lib_available = True
elif HAS_PDFPLUMBER:
pdf_lib = "pdfplumber"
pdf_lib_available = True
if not pdf_lib_available:
raise ToolError("PDF analysis requires PyMuPDF or pdfplumber")
try:
result = {
"success": False,
"file_path": file_path,
"processing_time": 0
}
if pdf_lib == "pymupdf":
# Use PyMuPDF for analysis
with pymupdf.open(file_path) as doc:
# Basic information
result["page_count"] = len(doc)
# Extract metadata if requested
if extract_metadata:
metadata = doc.metadata
if metadata:
result["metadata"] = {
"title": metadata.get("title", ""),
"author": metadata.get("author", ""),
"subject": metadata.get("subject", ""),
"keywords": metadata.get("keywords", ""),
"creator": metadata.get("creator", ""),
"producer": metadata.get("producer", ""),
"creation_date": metadata.get("creationDate", ""),
"modification_date": metadata.get("modDate", "")
}
# Extract outline if requested
if extract_outline:
toc = doc.get_toc()
if toc:
# Process TOC into a nested structure
result["outline"] = _process_toc(toc)
# Extract font information if requested
if extract_fonts:
fonts: Set[str] = set()
embedded_fonts: Set[str] = set()
for page_num in range(min(10, len(doc))): # Analyze first 10 pages
page = doc[page_num]
page_fonts = page.get_fonts()
for font in page_fonts:
fonts.add(font[3]) # Font name
if font[2]: # Embedded flag
embedded_fonts.add(font[3])
result["font_info"] = {
"total_fonts": len(fonts),
"embedded_fonts": len(embedded_fonts),
"font_names": list(fonts)
}
# Extract image information if requested
if extract_images:
image_count = 0
image_types: Dict[str, int] = {}
total_size = 0
for page_num in range(min(5, len(doc))): # Analyze first 5 pages
page = doc[page_num]
images = page.get_images(full=True)
for img in images:
image_count += 1
xref = img[0]
img_info = doc.extract_image(xref)
if img_info:
img_type = img_info["ext"]
img_size = len(img_info["image"])
image_types[img_type] = image_types.get(img_type, 0) + 1
total_size += img_size
# Extrapolate total images based on sample
estimated_total = int(image_count * (len(doc) / max(1, min(5, len(doc)))))
avg_size = f"{int(total_size / max(1, image_count) / 1024)}kb" if image_count > 0 else "0kb"
result["image_info"] = {
"total_images": image_count,
"estimated_total": estimated_total,
"image_types": image_types,
"average_size": avg_size
}
# Estimate OCR needs if requested
if estimate_ocr_needs:
text_pages = 0
total_pages = len(doc)
sample_size = min(10, total_pages)
for page_num in range(sample_size):
page = doc[page_num]
text = page.get_text()
if text and len(text.strip()) > 50: # Page has meaningful text
text_pages += 1
text_ratio = text_pages / sample_size
if text_ratio > 0.9:
needs_ocr = False
confidence = "high"
reason = "PDF contains extractable text throughout"
elif text_ratio > 0.5:
needs_ocr = True
confidence = "medium"
reason = "PDF has some extractable text but may benefit from OCR for certain pages"
else:
needs_ocr = True
confidence = "high"
reason = "PDF appears to be scanned or has minimal extractable text"
result["ocr_assessment"] = {
"needs_ocr": needs_ocr,
"confidence": confidence,
"reason": reason,
"text_coverage_ratio": text_ratio
}
elif pdf_lib == "pdfplumber":
# Use pdfplumber for analysis
with pdfplumber.open(file_path) as pdf:
# Basic information
result["page_count"] = len(pdf.pages)
# Extract metadata if requested
if extract_metadata:
metadata = pdf.metadata
if metadata:
result["metadata"] = {
"title": metadata.get("Title", ""),
"author": metadata.get("Author", ""),
"subject": metadata.get("Subject", ""),
"keywords": metadata.get("Keywords", ""),
"creator": metadata.get("Creator", ""),
"producer": metadata.get("Producer", ""),
"creation_date": metadata.get("CreationDate", ""),
"modification_date": metadata.get("ModDate", "")
}
# Outline not supported in pdfplumber
if extract_outline:
result["outline"] = []
# Font and image info not supported in pdfplumber
if extract_fonts:
result["font_info"] = {
"total_fonts": 0,
"embedded_fonts": 0,
"font_names": []
}
if extract_images:
result["image_info"] = {
"total_images": 0,
"image_types": {},
"average_size": "0kb"
}
# Estimate OCR needs if requested
if estimate_ocr_needs:
text_pages = 0
total_pages = len(pdf.pages)
sample_size = min(10, total_pages)
for page_num in range(sample_size):
page = pdf.pages[page_num]
text = page.extract_text()
if text and len(text.strip()) > 50: # Page has meaningful text
text_pages += 1
text_ratio = text_pages / sample_size
if text_ratio > 0.9:
needs_ocr = False
confidence = "high"
reason = "PDF contains extractable text throughout"
elif text_ratio > 0.5:
needs_ocr = True
confidence = "medium"
reason = "PDF has some extractable text but may benefit from OCR for certain pages"
else:
needs_ocr = True
confidence = "high"
reason = "PDF appears to be scanned or has minimal extractable text"
result["ocr_assessment"] = {
"needs_ocr": needs_ocr,
"confidence": confidence,
"reason": reason,
"text_coverage_ratio": text_ratio
}
# Update result
processing_time = time.time() - start_time
result["success"] = True
result["processing_time"] = processing_time
logger.info(f"PDF structure analysis completed in {processing_time:.2f}s")
return result
except Exception as e:
logger.error(f"Error in analyze_pdf_structure: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to analyze PDF structure: {str(e)}") from e
@with_tool_metrics
@with_retry(max_retries=2, retry_delay=1.0)
@with_error_handling
async def batch_process_documents(
folder_path: str,
file_pattern: str = "*.pdf",
output_folder: Optional[str] = None,
extraction_method: str = "hybrid",
max_pages_per_file: int = 0,
reformat_as_markdown: bool = True,
suppress_headers: bool = True,
max_concurrency: int = 3,
skip_on_error: bool = True,
bytes_data: Optional[Dict[str, Union[bytes, str]]] = None
) -> Dict[str, Any]:
"""
Processes multiple document files in a folder with OCR and LLM enhancement.
This tool handles batch processing of documents (PDFs and images) in a folder,
extracting text, correcting OCR errors, and saving the results to an output folder.
It can also process documents provided as bytes data.
Args:
folder_path: Path to the folder containing files to process
file_pattern: Pattern to match files (default: "*.pdf", can be "*.jpg", "*.png", etc.)
output_folder: Path to save the output files (default: create 'processed' subfolder)
extraction_method: Method for PDF text extraction ("direct", "ocr", "hybrid")
max_pages_per_file: Maximum pages to process per PDF (0 = all pages)
reformat_as_markdown: Whether to format the output as markdown (default: True)
suppress_headers: Whether to remove headers and footers (default: True)
max_concurrency: Maximum number of files to process in parallel (default: 3)
skip_on_error: Whether to continue processing other files if one fails (default: True)
bytes_data: Optional dictionary of filename to bytes data for processing data directly
Returns:
A dictionary containing:
{
"success": true,
"processed_files": [
{
"file": "/path/to/document1.pdf",
"output_file": "/path/to/output/document1.md",
"pages_processed": 5,
"extraction_method": "hybrid",
"processing_time": 12.34,
"quality_score": 85 # if quality assessment is performed
},
{
"file": "/path/to/document2.pdf",
"error": "Error message", # if processing failed
"status": "failed"
}
],
"total_files": 5,
"successful_files": 4,
"failed_files": 1,
"output_folder": "/path/to/output",
"total_processing_time": 45.67 # Seconds
}
Raises:
ToolInputError: If the folder path is invalid
ToolError: If batch processing fails
"""
start_time = time.time()
# Validate input if processing files from a folder
all_files = []
if not bytes_data:
# Standard file processing from a folder
if not folder_path or not os.path.exists(folder_path) or not os.path.isdir(folder_path):
raise ToolInputError(f"Invalid folder path: {folder_path}")
# Set output folder if not provided
if not output_folder:
output_folder = os.path.join(folder_path, "processed")
# Create output folder if it doesn't exist
os.makedirs(output_folder, exist_ok=True)
# Find files matching the pattern
matching_files: List[Path] = sorted(list(Path(folder_path).glob(file_pattern)))
if not matching_files:
raise ToolInputError(f"No files found in {folder_path} matching pattern {file_pattern}")
all_files = [(str(f), None) for f in matching_files] # (path, bytes_data)
else:
# Processing from bytes data
if not output_folder:
# Create a temporary output folder if not specified
output_folder = tempfile.mkdtemp(prefix="ocr_batch_")
else:
os.makedirs(output_folder, exist_ok=True)
# Convert bytes_data to our format
for filename, data in bytes_data.items():
if isinstance(data, str) and data.startswith('data:'):
# Handle base64 data URLs
try:
mime_type, b64data = data.split(';base64,', 1)
file_bytes = base64.b64decode(b64data)
all_files.append((filename, file_bytes))
except Exception as e:
logger.error(f"Error decoding base64 data for {filename}: {str(e)}")
if not skip_on_error:
raise ToolError(f"Failed to decode base64 data: {str(e)}") from e
elif isinstance(data, bytes):
# Already in bytes format
all_files.append((filename, data))
else:
logger.error(f"Unsupported data format for {filename}")
if not skip_on_error:
raise ToolInputError(f"Unsupported data format for {filename}")
if not all_files:
raise ToolInputError("No files to process")
# Get task type for batch processing
task_type = _get_task_type_for_ocr(extraction_method)
logger.info(f"Batch processing documents with task type: {task_type}")
# Initialize result
result = {
"success": False,
"processed_files": [],
"total_files": len(all_files),
"successful_files": 0,
"failed_files": 0,
"output_folder": output_folder,
"total_processing_time": 0,
"task_type": task_type
}
# Create semaphore for concurrency control
semaphore = asyncio.Semaphore(max_concurrency)
# Create partially-applied functions for better reuse and readability
# This allows us to pre-configure the processing functions with common parameters
extract_pdf_with_config = functools.partial(
extract_text_from_pdf,
extraction_method=extraction_method,
max_pages=max_pages_per_file,
skip_pages=0,
reformat_as_markdown=reformat_as_markdown,
suppress_headers=suppress_headers,
assess_quality=True
)
extract_pdf_bytes_with_config = functools.partial(
extract_text_from_pdf_bytes,
extraction_method=extraction_method,
max_pages=max_pages_per_file,
skip_pages=0,
reformat_as_markdown=reformat_as_markdown,
suppress_headers=suppress_headers,
assess_quality=True
)
process_image_with_config = functools.partial(
process_image_ocr,
reformat_as_markdown=reformat_as_markdown,
assess_quality=True
)
# Define worker function for processing each file
async def process_file(file_info: Tuple[str, Optional[bytes]]) -> Dict[str, Any]:
file_path, file_bytes = file_info
async with semaphore:
logger.info(f"Processing file: {file_path}")
file_start_time = time.time()
try:
# Determine file type based on extension
is_pdf = file_path.lower().endswith('.pdf')
# Process according to file type
if is_pdf:
# Extract base name
base_name = os.path.splitext(os.path.basename(file_path))[0]
# Determine output file extension
output_extension = '.md' if reformat_as_markdown else '.txt'
# Define output file path
output_file = os.path.join(output_folder, f"{base_name}{output_extension}")
# Extract text based on whether we have bytes or file path
if file_bytes is not None:
# Process PDF from bytes
extraction_result = await extract_pdf_bytes_with_config(pdf_bytes=file_bytes)
else:
# Process PDF from file path
extraction_result = await extract_pdf_with_config(file_path=file_path)
# Save the enhanced text
with open(output_file, "w", encoding="utf-8") as f:
f.write(extraction_result["text"])
# Save the raw text for reference
raw_output_file = os.path.join(output_folder, f"{base_name}_raw.txt")
with open(raw_output_file, "w", encoding="utf-8") as f:
f.write(extraction_result["raw_text"])
# Create file result
file_processing_time = time.time() - file_start_time
file_result = {
"file": file_path,
"output_file": output_file,
"raw_output_file": raw_output_file,
"pages_processed": extraction_result["pages_processed"],
"extraction_method_used": extraction_result["extraction_method_used"],
"processing_time": file_processing_time,
"status": "success"
}
# Add quality metrics if available
if "quality_metrics" in extraction_result:
quality_metrics = extraction_result["quality_metrics"]
file_result["quality_score"] = quality_metrics.get("score")
logger.info(f"Successfully processed PDF: {file_path}")
else:
# Handle image file
base_name = os.path.splitext(os.path.basename(file_path))[0]
output_extension = '.md' if reformat_as_markdown else '.txt'
output_file = os.path.join(output_folder, f"{base_name}{output_extension}")
# Process image with OCR based on whether we have bytes or file path
if file_bytes is not None:
# Process image from bytes
ocr_result = await process_image_with_config(image_data=base64.b64encode(file_bytes).decode('utf-8'))
else:
# Process image from file path
ocr_result = await process_image_with_config(image_path=file_path)
# Save the enhanced text
with open(output_file, "w", encoding="utf-8") as f:
f.write(ocr_result["text"])
# Save the raw text for reference
raw_output_file = os.path.join(output_folder, f"{base_name}_raw.txt")
with open(raw_output_file, "w", encoding="utf-8") as f:
f.write(ocr_result["raw_text"])
# Create file result
file_processing_time = time.time() - file_start_time
file_result = {
"file": file_path,
"output_file": output_file,
"raw_output_file": raw_output_file,
"table_detected": ocr_result.get("table_detected", False),
"processing_time": file_processing_time,
"status": "success"
}
# Add quality metrics if available
if "quality_metrics" in ocr_result:
quality_metrics = ocr_result["quality_metrics"]
file_result["quality_score"] = quality_metrics.get("score")
logger.info(f"Successfully processed image: {file_path}")
return file_result
except Exception as e:
logger.error(f"Error processing {file_path}: {str(e)}")
return {
"file": file_path,
"error": str(e),
"status": "failed"
}
try:
# Process files in parallel
tasks = [process_file(file_info) for file_info in all_files]
processed_results = await asyncio.gather(*tasks)
# Update result
result["processed_files"] = processed_results
result["successful_files"] = sum(1 for r in processed_results if r.get("status") == "success")
result["failed_files"] = sum(1 for r in processed_results if r.get("status") == "failed")
result["success"] = True
# Calculate total processing time
total_processing_time = time.time() - start_time
result["total_processing_time"] = total_processing_time
logger.info(f"Batch processing completed: {result['successful_files']} successful, {result['failed_files']} failed")
return result
except Exception as e:
logger.error(f"Error in batch processing: {str(e)}")
logger.error(traceback.format_exc())
raise ToolError(f"Failed to batch process documents: {str(e)}") from e
# --- Additional helper functions ---
def _process_toc(toc: List) -> List[Dict[str, Any]]:
"""
Processes a PDF table of contents into a nested structure.
Args:
toc: Table of contents from PyMuPDF
Returns:
Nested outline structure
"""
if not toc:
return []
# Convert flat list with indentation levels to nested structure
result = []
stack = [(-1, result)] # (level, children_list)
for item in toc:
level, title, page = item
# Find parent in stack
while stack[-1][0] >= level:
stack.pop()
# Create new entry
entry = {"title": title, "page": page, "children": []}
stack[-1][1].append(entry)
# Add to stack
stack.append((level, entry["children"]))
return result
async def _format_tables_in_text(text: str) -> str:
"""
Detects and formats potential tables in text using markdown.
Args:
text: Text to process
Returns:
Text with tables formatted in markdown
"""
# Simple pattern to detect table-like content
table_patterns = [
# Multiple lines with similar column separator patterns
r'(\n|^)(((\s*\S+\s*\|\s*\S+\s*)+\|?(\s*\n)){2,})',
# Multiple lines with similar tab/space alignment
r'(\n|^)((\s*\S+\s+\S+\s+\S+\s+\S+\s*\n){3,})'
]
table_sections: List[Tuple[int, int, str]] = []
for pattern in table_patterns:
matches = re.finditer(pattern, text, re.MULTILINE)
for match in matches:
table_sections.append((match.start(), match.end(), match.group(2)))
# Sort by start position
table_sections.sort(key=lambda x: x[0])
# No tables found
if not table_sections:
return text
# Process each potential table
result_parts = []
last_end = 0
for start, end, table_text in table_sections:
# Add text before table
if start > last_end:
result_parts.append(text[last_end:start])
# Process table
try:
formatted_table = await _enhance_table_formatting(table_text)
result_parts.append(formatted_table)
except Exception as e:
logger.warning(f"Error formatting table: {str(e)}")
result_parts.append(table_text)
last_end = end
# Add remaining text
if last_end < len(text):
result_parts.append(text[last_end:])
return ''.join(result_parts)
async def _enhance_table_formatting(table_text):
"""
Enhances table formatting using LLM.
Args:
table_text: Potential table text
Returns:
Formatted table in markdown
"""
prompt = f"""Format the following text as a markdown table. The text appears to contain tabular data but may not be properly formatted.
1. Detect column headers and content
2. Create a proper markdown table with headers, separator row, and content rows
3. Preserve all information but improve readability
4. If the input is not actually tabular data, return it unchanged with a comment indicating it's not a table
Here is the text to format:
```
{table_text}
```
Provide ONLY the formatted markdown table with no explanations or comments.
"""
try:
result = await generate_completion(
prompt=prompt,
provider=Provider.ANTHROPIC.value,
temperature=0.2,
max_tokens=len(table_text) + 500
)
if not result or not result.get("text"):
return table_text
formatted_table = result["text"]
# Check if it's actually formatted as a markdown table
if "|" in formatted_table and "-|-" in formatted_table:
return "\n" + formatted_table + "\n"
else:
return table_text
except Exception as e:
logger.warning(f"Error enhancing table format: {str(e)}")
return table_text
async def _assess_text_quality(original_text: str, enhanced_text: str) -> Dict[str, Any]:
"""
Assesses the quality of OCR enhancement using LLM.
Args:
original_text: Original OCR text
enhanced_text: LLM-enhanced text
Returns:
Dictionary with quality assessment
"""
# Truncate texts to reasonable lengths for assessment
max_sample = 5000
original_sample = original_text[:max_sample]
enhanced_sample = enhanced_text[:max_sample]
prompt = f"""Assess the quality improvement between the original OCR text and the enhanced version. Consider:
1. Error correction (typos, OCR artifacts, broken words)
2. Formatting improvements (paragraph structure, headings, lists)
3. Readability enhancement
4. Preservation of original content and meaning
5. Removal of unnecessary elements (headers, footers, artifacts)
Original OCR text:
```
{original_sample}
```
Enhanced text:
```
{enhanced_sample}
```
Provide:
1. A quality score from 0-100 where 100 is perfect enhancement
2. A brief explanation of improvements and any issues
3. Specific examples of corrections (max 3 examples)
Format your response as follows:
SCORE: [score]
EXPLANATION: [explanation]
EXAMPLES:
- [example 1]
- [example 2]
- [example 3]
"""
try:
result = await generate_completion(
prompt=prompt,
provider=Provider.ANTHROPIC.value,
temperature=0.3,
max_tokens=1000
)
if not result or not result.get("text"):
return {"score": None, "explanation": "Failed to assess quality"}
assessment_text = result["text"]
# Parse the assessment
score_match = re.search(r'SCORE:\s*(\d+)', assessment_text)
explanation_match = re.search(r'EXPLANATION:\s*(.*?)(?:\n\s*EXAMPLES|\Z)', assessment_text, re.DOTALL)
examples_match = re.search(r'EXAMPLES:\s*(.*?)(?:\Z)', assessment_text, re.DOTALL)
score = int(score_match.group(1)) if score_match else None
explanation = explanation_match.group(1).strip() if explanation_match else "No explanation provided"
examples = []
if examples_match:
examples_text = examples_match.group(1)
examples = [ex.strip().lstrip('- ') for ex in examples_text.split('\n') if ex.strip()]
return {
"score": score,
"explanation": explanation,
"examples": examples
}
except Exception as e:
logger.warning(f"Error assessing text quality: {str(e)}")
return {"score": None, "explanation": f"Failed to assess quality: {str(e)}"}
```