extract_from_url

Instructions

Implementation Reference

• src/langextract_mcp/server.py:319-402 (handler)

  The primary handler function for the 'extract_from_url' MCP tool. Decorated with @mcp.tool for automatic registration. Validates URL and parameters, creates ExtractionConfig, retrieves API key, calls _langextract_client.extract(url), and returns formatted results using _format_extraction_result.

  @mcp.tool
  def extract_from_url(
      url: str,
      prompt_description: str,
      examples: list[dict[str, Any]],
      model_id: str = "gemini-2.5-flash",
      max_char_buffer: int = 1000,
      temperature: float = 0.5,
      extraction_passes: int = 1,
      max_workers: int = 10
  ) -> dict[str, Any]:
      """
      Extract structured information from text content at a URL.
      
      Downloads text from the specified URL and extracts structured information
      using Large Language Models. Ideal for processing web articles, documents,
      or any text content accessible via HTTP/HTTPS.
      
      Args:
          url: URL to download text from (must start with http:// or https://)
          prompt_description: Clear instructions for what to extract
          examples: List of example extractions to guide the model
          model_id: LLM model to use (default: "gemini-2.5-flash")
          max_char_buffer: Max characters per chunk (default: 1000)
          temperature: Sampling temperature 0.0-1.0 (default: 0.5)
          extraction_passes: Number of extraction passes for better recall (default: 1)
          max_workers: Max parallel workers (default: 10)
          
      Returns:
          Dictionary containing extracted entities with source locations and metadata
          
      Raises:
          ToolError: If URL is invalid, download fails, or extraction fails
      """
      try:
          if not url.startswith(('http://', 'https://')):
              raise ToolError("URL must start with http:// or https://")
              
          if not examples:
              raise ToolError("At least one example is required for reliable extraction")
          
          if not prompt_description.strip():
              raise ToolError("Prompt description cannot be empty")
          
          # Validate that only Gemini models are supported
          if not model_id.startswith('gemini'):
              raise ToolError(
                  f"Only Google Gemini models are supported. Got: {model_id}. "
                  f"Use 'list_supported_models' tool to see available options."
              )
          
          # Create config object from individual parameters
          config = ExtractionConfig(
              model_id=model_id,
              max_char_buffer=max_char_buffer,
              temperature=temperature,
              extraction_passes=extraction_passes,
              max_workers=max_workers
          )
          
          # Get API key (server-side only for security)
          api_key = _get_api_key()
          if not api_key:
              raise ToolError(
                  "API key required. Server administrator must set LANGEXTRACT_API_KEY environment variable."
              )
          
          # Perform optimized extraction using cached client
          result = _langextract_client.extract(
              text_or_url=url,
              prompt_description=prompt_description,
              examples=examples,
              config=config,
              api_key=api_key
          )
          
          return _format_extraction_result(result, config, source_url=url)
          
      except ValueError as e:
          raise ToolError(f"Invalid parameters: {str(e)}")
      except Exception as e:
          raise ToolError(f"URL extraction failed: {str(e)}")

• src/langextract_mcp/server.py:21-28 (schema)

  Pydantic ExtractionConfig model defining input parameters for the extraction process, used by both extract_from_text and extract_from_url tools.

  class ExtractionConfig(BaseModel):
      """Configuration for extraction parameters."""
      model_id: str = Field(default="gemini-2.5-flash", description="LLM model to use")
      max_char_buffer: int = Field(default=1000, description="Max characters per chunk")
      temperature: float = Field(default=0.5, description="Sampling temperature (0.0-1.0)")
      extraction_passes: int = Field(default=1, description="Number of extraction passes for better recall")
      max_workers: int = Field(default=10, description="Max parallel workers")

• src/langextract_mcp/server.py:38-190 (helper)

  LangExtractClient class implementing the core extraction logic delegated to by the tool handlers. Provides caching for language models, schemas, resolvers, and performs the actual annotation using langextract library.

  class LangExtractClient:
      """Optimized langextract client for MCP server usage.
      
      This client maintains persistent connections and caches expensive operations
      like schema generation and prompt templates for better performance in a
      long-running MCP server context.
      """
      
      def __init__(self):
          self._language_models: dict[str, Any] = {}
          self._schema_cache: dict[str, Any] = {}
          self._prompt_template_cache: dict[str, Any] = {}
          self._resolver_cache: dict[str, Any] = {}
          
      def _get_examples_hash(self, examples: list[dict[str, Any]]) -> str:
          """Generate a hash for caching based on examples."""
          examples_str = json.dumps(examples, sort_keys=True)
          return hashlib.md5(examples_str.encode()).hexdigest()
      
      def _get_language_model(self, config: ExtractionConfig, api_key: str, schema: Any | None = None, schema_hash: str | None = None) -> Any:
          """Get or create a cached language model instance."""
          # Include schema hash in cache key to prevent schema mutation conflicts
          model_key = f"{config.model_id}_{config.temperature}_{config.max_workers}_{schema_hash or 'no_schema'}"
          
          if model_key not in self._language_models:
              # Validate that only Gemini models are supported
              if not config.model_id.startswith('gemini'):
                  raise ValueError(f"Only Gemini models are supported. Got: {config.model_id}")
                  
              language_model = lx.inference.GeminiLanguageModel(
                  model_id=config.model_id,
                  api_key=api_key,
                  temperature=config.temperature,
                  max_workers=config.max_workers,
                  gemini_schema=schema
              )
              self._language_models[model_key] = language_model
              
          return self._language_models[model_key]
      
      def _get_schema(self, examples: list[dict[str, Any]], model_id: str) -> tuple[Any, str]:
          """Get or create a cached schema for the examples.
          
          Returns:
              Tuple of (schema, examples_hash) for use in caching language models
          """
          if not model_id.startswith('gemini'):
              return None, ""
              
          examples_hash = self._get_examples_hash(examples)
          schema_key = f"{model_id}_{examples_hash}"
          
          if schema_key not in self._schema_cache:
              # Convert examples to langextract format
              langextract_examples = self._create_langextract_examples(examples)
              
              # Create prompt template to generate schema
              prompt_template = lx.prompting.PromptTemplateStructured(description="Schema generation")
              prompt_template.examples.extend(langextract_examples)
              
              # Generate schema
              schema = lx.schema.GeminiSchema.from_examples(prompt_template.examples)
              self._schema_cache[schema_key] = schema
              
          return self._schema_cache[schema_key], examples_hash
      
      def _get_resolver(self, format_type: str = "JSON") -> Any:
          """Get or create a cached resolver."""
          if format_type not in self._resolver_cache:
              resolver = lx.resolver.Resolver(
                  fence_output=False,
                  format_type=lx.data.FormatType.JSON if format_type == "JSON" else lx.data.FormatType.YAML,
                  extraction_attributes_suffix="_attributes",
                  extraction_index_suffix=None,
              )
              self._resolver_cache[format_type] = resolver
              
          return self._resolver_cache[format_type]
      
      def _create_langextract_examples(self, examples: list[dict[str, Any]]) -> list[lx.data.ExampleData]:
          """Convert dictionary examples to langextract ExampleData objects."""
          langextract_examples = []
          
          for example in examples:
              extractions = []
              for extraction_data in example["extractions"]:
                  extractions.append(
                      lx.data.Extraction(
                          extraction_class=extraction_data["extraction_class"],
                          extraction_text=extraction_data["extraction_text"],
                          attributes=extraction_data.get("attributes", {})
                      )
                  )
              
              langextract_examples.append(
                  lx.data.ExampleData(
                      text=example["text"],
                      extractions=extractions
                  )
              )
          
          return langextract_examples
      
      def extract(
          self, 
          text_or_url: str,
          prompt_description: str,
          examples: list[dict[str, Any]],
          config: ExtractionConfig,
          api_key: str
      ) -> lx.data.AnnotatedDocument:
          """Optimized extraction using cached components."""
          # Get or generate schema first
          schema, examples_hash = self._get_schema(examples, config.model_id)
          
          # Get cached components with schema-aware caching
          language_model = self._get_language_model(config, api_key, schema, examples_hash)
          resolver = self._get_resolver("JSON")
          
          # Convert examples
          langextract_examples = self._create_langextract_examples(examples)
          
          # Create prompt template
          prompt_template = lx.prompting.PromptTemplateStructured(
              description=prompt_description
          )
          prompt_template.examples.extend(langextract_examples)
          
          # Create annotator
          annotator = lx.annotation.Annotator(
              language_model=language_model,
              prompt_template=prompt_template,
              format_type=lx.data.FormatType.JSON,
              fence_output=False,
          )
          
          # Perform extraction
          if text_or_url.startswith(('http://', 'https://')):
              # Download text first
              text = lx.io.download_text_from_url(text_or_url)
          else:
              text = text_or_url
              
          return annotator.annotate_text(
              text=text,
              resolver=resolver,
              max_char_buffer=config.max_char_buffer,
              batch_length=10,
              additional_context=None,
              debug=False,  # Disable debug for cleaner MCP output
              extraction_passes=config.extraction_passes,
          )

• src/langextract_mcp/server.py:201-230 (helper)

  _format_extraction_result helper function that converts the langextract AnnotatedDocument result into the dictionary format returned by the tool.

  def _format_extraction_result(result: lx.data.AnnotatedDocument, config: ExtractionConfig, source_url: str | None = None) -> dict[str, Any]:
      """Format langextract result for MCP response."""
      extractions = []
      
      for extraction in result.extractions or []:
          extractions.append({
              "extraction_class": extraction.extraction_class,
              "extraction_text": extraction.extraction_text,
              "attributes": extraction.attributes,
              "start_char": getattr(extraction, 'start_char', None),
              "end_char": getattr(extraction, 'end_char', None),
          })
      
      response = {
          "document_id": result.document_id if result.document_id else "anonymous",
          "total_extractions": len(extractions),
          "extractions": extractions,
          "metadata": {
              "model_id": config.model_id,
              "extraction_passes": config.extraction_passes,
              "max_char_buffer": config.max_char_buffer,
              "temperature": config.temperature,
          }
      }
      
      if source_url:
          response["source_url"] = source_url
          
      return response

• src/langextract_mcp/server.py:196-199 (helper)

  _get_api_key helper function that retrieves the required Google Gemini API key from environment variable.

  def _get_api_key() -> str | None:
      """Get API key from environment (server-side only for security)."""
      return os.environ.get("LANGEXTRACT_API_KEY")