server.py

"""
STRING MCP Server
=================

This script exposes STRING database functionality as an MCP (Model Context Protocol) server.
It provides tools for resolving protein identifiers, retrieving interactions, functional
annotations, enrichment analysis, and generating network visualizations.

The server is implemented in Python.

Configuration:
  - Reads settings from config/server.config (JSON)
  - Requires "base_url" (STRING API endpoint) and "server_port"

Run:
  python server.py

The server listens on the configured port and serves tools via streamable-http transport.

Requirements:

fastmcp>=2.10.6,<2.11
httpx>=0.28,<0.29
pydantic>=2.11,<2.12

Creator: meringlab
Contact email: damian.szklarczyk@sib.swiss 
License: CC-BY-4.0

"""

import sys
import json
import time
import httpx
import asyncio
import traceback

from collections import defaultdict
from typing import Annotated, Optional

from pydantic import Field
from fastmcp import FastMCP

from string_help import HELP_TOPICS

try:
    with open("config/server.config") as f:
        config = json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
    sys.stderr.write(f"Error loading config: {e}\n")
    sys.exit(1)


base_url = config.get("base_url")
server_port = int(config.get("server_port", 0))


if not base_url:
    raise ValueError("Missing required config: 'base_url', e.g. 'https://version-12-0.string-db.org' ")

if not server_port:
    raise ValueError("Missing required config: 'server_port', e.g. '57416' ")

timeout = float(config.get("timeout", 100))


def _normalize_required_score(required_score: Optional[int]) -> Optional[int]:
    if required_score is None:
        return None

    if isinstance(required_score, str):
        required_score = required_score.strip()
        if required_score.lower() in ("", "null", "none"):
            return 0

    try:
        return int(required_score)
    except (TypeError, ValueError):
        return 0


## logging verbosity ## 

log_verbosity = {}
log_verbosity['call'] = False
log_verbosity['params'] = False
log_verbosity["size"] =  False


if 'verbosity' in config:

    if config['verbosity'] == 'full':
        log_verbosity['call'] = True
        log_verbosity['params'] = True
        log_verbosity['size'] = True

    if config['verbosity'] == 'low':
        log_verbosity['call'] = True
        log_verbosity['params'] = False
        log_verbosity['size'] = True


async def _post_json(client: httpx.AsyncClient, endpoint: str, data: dict):
    """
    POST form data to a STRING API endpoint and return JSON.
    Emits periodic 'ping' SSE events to keep upstream connections alive (e.g. Cloudflare).
    """
    params = data
    ping_interval = 25  # seconds between pings

    async def _ping_loop(done_event: asyncio.Event):
        try:
            while not done_event.is_set():
                await asyncio.sleep(ping_interval)
                if not done_event.is_set():
                    print(json.dumps({"type": "ping", "message": "waiting..."}), flush=True)
                    print("[_post_json] ping (still waiting for response)", file=sys.stderr, flush=True)
        except asyncio.CancelledError:
            pass

    done_event = asyncio.Event()
    ping_task = asyncio.create_task(_ping_loop(done_event))

    try:
        response = await client.post(endpoint, data=params)
        response.raise_for_status()
        try:
            return response.json()
        except ValueError:
            return {"result": response.text}

    except httpx.ReadTimeout:
        # Specific handling for slow or non-responsive STRING API calls
        sys.stderr.write(f"[timeout] STRING API request timed out at {endpoint}\n")
        error_payload = {
            "error": {
                "type": "timeout_error",
                "message": f"STRING API request to {endpoint} timed out.",
                "hints": [
                    "Try narrowing the taxon",
                    "Limiting the number of proteins",
                    "Requery the API again",
                ],
                "diagnostics": {"params_sent": {k: v for k, v in params.items()}},
            }
        }
        sys.stderr.write(
            f"[handled] Returned payload after timeout: {json.dumps(error_payload, ensure_ascii=False)}\n"
        )
        return error_payload

    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        try:
            server_detail = e.response.json()
        except Exception:
            server_detail = e.response.text

        diagnostics = {
            "endpoint": endpoint,
            "status": status,
            "params_sent": {k: v for k, v in params.items()},
            "server_detail": server_detail,
        }

        hints = []
        if status == 400:
            if not params.get("species") or not params.get("species").isnumeric():
                hints.append(
                    "Valid 'species' parameter required (NCBI taxonomy ID or clade). "
                    "Example: 9606 (human), 7227 (D. melanogaster), 10090 (mouse), or a STRING clade ID. "
                )
            elif not params.get("identifiers"):
                hints.append("Missing 'identifiers'.")
            else:
                hints.append(
                    "Species might be not present in STRING. Search string_query_species with a name "
                    "or invoke 'string_help' with topic='missing_species'. "
                    "For multiple-protein queries make sure the delimiter is '%0d'. "
                )
        elif status == 404:
            hints.append(
                "The provided identifiers could not be mapped in STRING. "
                "Use 'string_resolve_proteins' first to resolve ambiguous names. "
                "For multiple-protein queries make sure the delimiter is '%0d'. "
            )

        error_payload = {
            "error": {
                "type": "string_api_error",
                "message": f"STRING API request failed with status {status}.",
                "hints": hints or ["Check parameters and retry."],
                "diagnostics": diagnostics,
            }
        }

        sys.stderr.write(
            f"[handled] HTTPStatusError {status} at {endpoint} with params={params}\n"
            f"[handled] Returned payload: {json.dumps(error_payload, ensure_ascii=False)}\n"
        )
        return error_payload

    except Exception as e:
        traceback.print_exc(file=sys.stderr)
        error_payload = {
            "error": {
                "type": "unexpected_error",
                "message": "Unexpected error in STRING API call.",
                "diagnostics": {"params_sent": {k: v for k, v in params.items()}},
            }
        }
        sys.stderr.write(
            f"[handled] Unexpected exception {type(e).__name__} at {endpoint} with params={params}\n"
            f"[handled] Returned payload: {json.dumps(error_payload, ensure_ascii=False)}\n"
        )
        return error_payload

    finally:
        # stop the ping loop cleanly
        done_event.set()
        ping_task.cancel()


mcp = FastMCP(
    name="STRING Database MCP Server",
)

@mcp.tool(title="STRING: Resolves protein identifiers to metadata")
async def string_resolve_proteins(
    proteins: Annotated[
        str,
        Field(
            description=(
                "Required. One or more input protein identifiers (gene symbols, UniProt IDs, etc.), "
                "separated by carriage return (%0d). Example: TP53%0dSMO"
            )
        )
    ],
    species: Annotated[
        str,
        Field(
            description=(
                "Optional. NCBI taxonomy ID (e.g. 9606 for human) or STRING genome ID "
                "(e.g. STRG0AXXXXX for uploaded genomes)."
            )
        )
    ] = None,
    show_sequence: Annotated[
        str,
        Field(
            description=(
                "Optional. '1' to include sequences (default '0'). Use only if the user requests sequence data."
            )
        )
    ] = None
) -> dict:
    """
    Maps one or more protein identifiers to their corresponding STRING metadata, including:
    gene symbol, description, sequence, domains, species, and internal STRING ID.

    This method is useful for translating raw identifiers into readable, annotated protein entries.

    Example input: "TP53%0dSMO"
    """

   
    params = {"identifiers": proteins, "echo_query": 1, 'add_domains': 1}
    if species is not None:
        params["species"] = species

    if show_sequence is not None:
        params["add_sequence"] = show_sequence

    endpoint = "/api/json/get_string_ids"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        log_response_size(results)
        return {"mapped_proteins": results}


@mcp.tool(title="STRING: Get interactions within query set")
async def string_interactions_query_set(
    proteins: Annotated[
        str,
        Field(description=(
            "Required. One or more protein identifiers, separated by carriage return (%0d). "
            "Example: SMO%0dTP53"
        ))
    ],
    species: Annotated[
        str,
        Field(description=(
            "Optional. NCBI taxonomy ID (e.g. 9606 for human) or STRING genome ID "
            "(e.g. STRG0AXXXXX for uploaded genomes)."
        ))
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description=(
            "Optional. Minimum confidence score for an interaction (range: 0–1000). "
            "Only set this if the user explicitly requests it."
        ))
    ] = None,
    network_type: Annotated[
        Optional[str],
        Field(description=(
            'Optional. Network type: "functional" (default) or "physical" (co-complex).'
        ))
    ] = None,
    extend_network: Annotated[
        Optional[int],
        Field(description=(
            "Optional. Number of additional proteins to add to the network based on their "
            "connectivity. Default is 10 for a single protein query and 0 for multiple proteins. "
            "Only set this if the user explicitly requests it."
        ))
    ] = None,
    #show_query_node_labels: Annotated[
    #    Optional[int],
    #    Field(description=(
    #        "Optional. Set to 1 to display user-supplied names instead of STRING preferred names "
    #        "in the output. Default is 0. Only set if the user explicitly requests it."
    #    ))
    #] = None
) -> dict:
    """
    Retrieves the interactions between the query proteins.
    Use this method only when you specifically need to list the interactions between all proteins in your query set.
    If user asks for 'physical' or 'complex' use 'physical' network type.
    
    - For a **single protein**, the network includes that protein and its top 10 most likely interaction partners, plus all interactions among those partners.
    - For **multiple proteins**, the network includes all direct interactions between them.
    - If the user refers to "physical interactions", "complexes", or "binding", set the network type to "physical".
    - STRING does not store or report information about self-interactions/homomers; if asked, explain the limitation.
    
    If few or no interactions are returned, consider reducing the `required_score`.
    
    For large query sets (>50 proteins), consider increasing the `required_score` (e.g. ≥700) 
    to focus on high-confidence interactions and avoid overly dense networks.
    
    - Expand the names of score sources:  
        `nscore` (neighborhood), `fscore` (fusion), `pscore` (phylogenetic profile),  
        `ascore` (coexpression), `escore` (experimental), `dscore` (database), `tscore` (text-mining)
    """

    required_score = _normalize_required_score(required_score)

    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species
    if required_score is not None:
        params["required_score"] = required_score
    if network_type is not None:
        params["network_type"] = network_type
    if extend_network is not None:
        params["add_white_nodes"] = extend_network
    #if show_query_node_labels is not None:
    #    params["show_query_node_labels"] = show_query_node_labels

    add_score_note = False
    add_shared_note = False

    if not required_score and len(proteins.lower().split("%0d")) <= 5:
        params["required_score"] = 0
        required_score = 0
        add_score_note = True
 
    if len(proteins.lower().split('%0d')) in [2,3] and extend_network is None:
        params['add_white_nodes'] = 5
        add_shared_note = True

    endpoint = "/api/json/network"

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results: return results
        else:
            formatted_network = format_query_set_network(results, required_score)

        notes = []


        if add_score_note:
            notes.append("The required_score parameter was lowered to 0 - showing all interactions. "
                         "IMPORTANT: If the interaction score is low (below 400), inform user about it.")
        if not len(formatted_network["network"]):
             notes.append(f"No interactions found in STRING database at that {required_score} cut-off.")

        notes.extend(formatted_network["notes"])
     
        if add_shared_note:
            notes.append(
                "For two- and three-protein queries, the network was expanded by five additional proteins to reveal possible shared or indirect interactions between the queried proteins. "
                "Report only interactions that are directly between the queried proteins or that form a clear shared-interactor/indirect path connecting the queried proteins. "
                "Do not mention expanded-network edges that are unrelated to the relationship between the queried proteins. "
                "For a more thorough investigation of shared interactors, use the `string_all_interaction_partners` tool."
            )

        response = {
            "notes": notes,
            "network_summary": formatted_network["network_summary"],
            "network": formatted_network["network"],
        }

        if formatted_network["node_interaction_counts"]:
            response["node_interaction_counts"] = formatted_network["node_interaction_counts"]

        if formatted_network["edge_sample"]:
            response["edge_sample"] = formatted_network["edge_sample"]

        log_response_size(response)

        return response


@mcp.tool(title="STRING: Get all interaction partners for proteins")
async def string_all_interaction_partners(
    identifiers: Annotated[
        str,
        Field(description=(
            "Required. One or more protein identifiers, separated by carriage return (%0d). "
            "Example: TP53%0dSMO"
        ))
    ],
    species: Annotated[
        str,
        Field(description=(
            "Optional. NCBI taxonomy ID (e.g. 9606 for human) or STRING genome ID "
            "(e.g. STRG0AXXXXX for uploaded genomes). Only set if the user explicitly requests it."
        ))
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description=(
            "Optional. Minimum interaction score to include (range: 0–1000). "
            "Only set if the user explicitly requests it."
        ))
    ] = None,
    network_type: Annotated[
        Optional[str],
        Field(description=(
            'Optional. Network type: "functional" (default) or "physical" (co-complex).'
        ))
    ] = None
) -> dict:
    """
    Retrieves all interaction partners for one or more proteins from STRING.

    This tool returns all known interactions between your query protein(s) and **any other proteins in the STRING database**.
    
    - Use this when asking **“What does TP53 interact with?”**
    - It differs from the `network` tool, which only shows interactions **within the input set** or a limited extension of it.
    - If the user refers to "physical interactions", "complexes", or "binding", set the network type to "physical".

    You can filter for strong interactions using `required_score`.

    - Evidence scores:  
        `nscore` (neighborhood), `fscore` (fusion), `pscore` (phylogenetic profile),  
        `ascore` (coexpression), `escore` (experimental), `dscore` (database), `tscore` (text mining)
    """

    required_score = _normalize_required_score(required_score)

    params = {"identifiers": identifiers, "limit": 0}

    if species is not None:
        params["species"] = species
    if required_score is not None:
        params["required_score"] = required_score
    if network_type is not None:
        params["network_type"] = network_type

    endpoint = "/api/json/interaction_partners"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results:
            log_response_size(results)
            return results
        else:
            formatted_interactions = format_interaction_partners(results, required_score)

        notes = []

        notes.extend(formatted_interactions["notes"])

        if not len(formatted_interactions["interactions"]):
             notes.append(f"No interactions found in STRING database at that {required_score} cut-off. Consider lowering the required_score.")

        response = {
            "notes": notes,
            "node_summary": formatted_interactions["node_summary"],
            "interactions": formatted_interactions["interactions"],
        }
        
        log_response_size(response)
        return response


@mcp.tool(title="STRING: Get interaction network image (image URL)")
async def string_visual_network(
    proteins: Annotated[
        str,
        Field(description="Required. One or more protein IDs, optionally followed by one numeric value per protein. Example:\n"
                  "PTEN 0.234\nSMO -3.445\n"
                  "Use newline (%0d) between entries. Tabs and spaces are accepted as separators.")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX).")
    ] = None,
    extend_network: Annotated[
        Optional[int],
        Field(description="Optional. Add specified number of nodes to the network, based on their scores (default: 0, or 10 for single protein queries).")
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description="Optional. Threshold of significance to include an interaction (0-1000). Default: 400. Increase for large queries to 700.")
    ] = None,
    network_type: Annotated[
        Optional[str],
        Field(description='Optional. Network type: "functional" (default) or "physical" (co-complex).')
    ] = None,
    network_flavor: Annotated[
        Optional[str],
        Field(description='Optional. Edge style: "evidence" (default), "confidence" (recommended for large queries, e.g. >100 proteins), or "actions".')
    ] = None,
    hide_disconnected_nodes: Annotated[
        Optional[int],
        Field(description="Optional. 1 to hide proteins not connected to any other protein, 0 otherwise (default: 0). DO NOT SET unless user explicitly requests.")
    ] = None,
    #show_query_node_labels: Annotated[
    #    Optional[int],
    #    Field(description="Optional. 1 display the user's query name(s) instead of STRING preferred name, (default: 0). DO NOT SET unless user explicitly requests.")
    #] = None,
    center_node_labels: Annotated[
        Optional[int],
        Field(description="Optional. 1 to center protein names on nodes, 0 otherwise (default: 0). DO NOT SET unless user explicitly requests.")
    ] = None,
    do_not_show_structures: Annotated[
        Optional[int],
        Field(description="Optional. 1 remove small protein structure previews from inside the node bubbles. DO NOT SET unless user explicitly requests.")
    ] = None,
 
    #custom_label_font_size: Annotated[
    #    Optional[int],
    #    Field(description="Optional. Change font size of protein names (from 5 to 50, default: 12). DO NOT SET unless user explicitly requests.")
    #] = None
) -> dict:
    """
    Retrieves a URL to a **STRING interaction network image** for one or more proteins.
    
    - For a single protein: includes the protein and its top 10 most likely interactors.
    - For multiple proteins: includes all known interactions **within the query set**.
    - If the user asks for "physical interactions", "complexes", or "binding", set `network_type` to "physical".
    
    The input may include one numeric value per protein, such as fold change, effect size, or score.
    These values are visualized as colored halos around the nodes, allowing overlay of protein-level measurements on the network.
    
    Example:
    PTEN 2.1
    SMO -1.3
    
    If numeric values are provided:
    - positive values are shown in blue
    - negative values are shown in red
    - larger absolute values produce stronger halo intensity
    
    If the user provides numeric values together with the proteins, preserve them in the query.
    
    If few or no interactions are shown, consider lowering `required_score`.
    
    For large queries (>100 proteins):
    - use `network_flavor="confidence"`
    - increase `required_score` (e.g. 700)
    
    Always ask if the user also wants a link to the interactive STRING network page.
    
    Input parameters should match those used in related STRING tools (e.g. `string_interactions_query_set`), unless otherwise specified.

    """
    required_score = _normalize_required_score(required_score)

    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species
    if extend_network is not None:
        params["add_white_nodes"] = extend_network
    if required_score is not None:
        params["required_score"] = required_score
    if network_type is not None:
        params["network_type"] = network_type
    if network_flavor is not None:
        params["network_flavor"] = network_flavor
    if hide_disconnected_nodes is not None:
        params["hide_disconnected_nodes"] = hide_disconnected_nodes
    if do_not_show_structures is not None:
        params["block_structure_pics_in_bubbles"] = do_not_show_structures
    #if show_query_node_labels is not None:
    #    params["show_query_node_labels"] = show_query_node_labels
    if center_node_labels is not None:
        params["center_node_labels"] = center_node_labels
    #if custom_label_font_size is not None:
    #    params["custom_label_font_size"] = custom_label_font_size

    endpoint = f"/api/json/network_image_url"

    add_score_note = False
    add_shared_note = False

    if not required_score and len(proteins.lower().split("%0d")) <= 5:
        params["required_score"] = 0
        add_score_note = True
   
    if len(proteins.lower().split('%0d')) in [2,3] and extend_network is None:
        params['add_white_nodes'] = 5
        add_shared_note = True


    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)

        notes = []

        if add_score_note:
            notes.append("For small queries, the `required_score` parameter was lowered to 0.")
        notes.append("Internal guidance: do not repeat this note to the user. The user can inspect the generated image, "
                     "but you should not infer or report specific interaction details from the image alone. "
                     "Use `string_interactions_query_set` with the same parameters to verify direct or indirect interactions.")

        if add_shared_note:
            notes.append(
                "For two- and three-protein queries, the network image was expanded by five additional proteins to reveal possible shared or indirect interactions between the queried proteins. "
                "Do not infer interactions from the image alone; use `string_interactions_query_set` with the same parameters to verify whether direct or indirect interactions are present. "
                "Report only interactions that are directly between the queried proteins or that form a clear shared-interactor/indirect path connecting the queried proteins. "
                "Do not mention expanded-network edges that are unrelated to the relationship between the queried proteins. "
                "For a more thorough investigation of shared interactors, use the `string_all_interaction_partners` tool.")

        notes.append("Embed the returned image link directly in the assistant response as a markdown image.")

 
        log_response_size(results)

        return {"notes": notes, "image_url": results}


@mcp.tool(title="STRING: Perform network clustering")
async def string_network_clustering(
    proteins: Annotated[
        str,
        Field(description=(
            "Required. One or more protein identifiers (optionally with values). Example:\n"
            "PTEN 0.234\nSMO -3.445\n"
            "Separate entries with newline (%0d). "
            "Numeric values (e.g. expression data) can be provided after identifiers."
        ))
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxonomy ID (e.g. 9606 for human, or STRG0AXXXXX for uploaded genomes).")
    ] = None,
    extend_network: Annotated[
        Optional[int],
        Field(description="Optional. Add specified number of additional nodes to the network based on their interaction scores (default: 0, or 10 for single-protein queries).")
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description="Optional. Minimum interaction confidence score (0–1000). Lower values show more edges. Default: 400.")
    ] = None,
    network_type: Annotated[
        Optional[str],
        Field(description='Optional. Network type: "functional" (default) or "physical" (co-complex).')
    ] = None,
    clustering_algorithm: Annotated[
        Optional[str],
        Field(description=(
            "Optional. Clustering algorithm: 'MCL' or 'kmeans'.\n"
            "- 'MCL' (Markov Cluster Algorithm) identifies densely connected subnetworks based on connectivity flow; "
            "- 'kmeans' partitions proteins into a fixed number of clusters, useful when explicit cluster counts are desired; "
       ))
    ] = None,
    
    clustering_parameter: Annotated[
        Optional[float],
        Field(description=(
            "Optional. Controls the clustering granularity:\n"
            "- For 'MCL': inflation parameter (1.0–10.0, default 3.0); higher values produce more, smaller clusters.\n"
            "- For 'kmeans': number of clusters (integer ≥2, default 3).\n"
        ))
    ] = None,
    network_flavor: Annotated[
        Optional[str],
        Field(description='Optional. Edge display style: "evidence" (default), "confidence", or "actions".')
    ] = None,
    hide_disconnected_nodes: Annotated[
        Optional[int],
        Field(description="Optional. 1 to hide unconnected nodes, 0 otherwise (default: 0). Use only if user explicitly requests it.")
    ] = None,
    center_node_labels: Annotated[
        Optional[int],
        Field(description="Optional. 1 to center protein labels on nodes, 0 otherwise (default: 0). Use only if user explicitly requests it.")
    ] = None,
) -> dict:
    """
    Performs **network clustering** on a STRING interaction network and returns both a **network image URL**
    and details about each detected cluster.
    
    Use the same parameters as in the network creation step to ensure consistency.
    If the network already contains disconnected subgraphs, the resulting number of clusters may differ from the requested value.
    
    Dashed lines represent connections between clusters, while solid lines indicate interactions within clusters.
    
    Notes:
      - For small queries (≤5 proteins), the `required_score` parameter is automatically lowered to 0.
      - If only a single cluster is produced, try increasing `required_score`, adjusting the inflation parameter,
        or switching to `kmeans` for small, highly interconnected networks.


    """

    required_score = _normalize_required_score(required_score)

    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species
    if extend_network is not None:
        params["add_white_nodes"] = extend_network
    if required_score is not None:
        params["required_score"] = required_score
    if network_type is not None:
        params["network_type"] = network_type
    if network_flavor is not None:
        params["network_flavor"] = network_flavor
    if hide_disconnected_nodes is not None:
        params["hide_disconnected_nodes"] = hide_disconnected_nodes


    # default
    if not clustering_algorithm:
        clustering_algorithm = 'MCL'

    # fix casing

    if clustering_algorithm.lower() == 'kmeans': clustering_algorithm = 'kmeans'
    if clustering_algorithm.lower() == 'mcl': clustering_algorithm = 'MCL'

    if clustering_algorithm not in ['MCL', 'kmeans']:
        clustering_algorithm = 'MCL'

    params['network_clustering_algorithm'] = clustering_algorithm

    # parse parameters

    if clustering_algorithm == 'kmeans':

        try:
            clustering_parameter = float(int(clustering_parameter))
        except Exception:
            clustering_parameter = 3

        params['network_clustering_parameter_kmeans'] = clustering_parameter

    if clustering_algorithm == 'MCL':

        try:
            clustering_parameter = float(clustering_parameter)
            clustering_parameter = max(1, min(10, clustering_parameter))
        except Exception:
            clustering_parameter = 3

        params['network_clustering_parameter_mcl'] = clustering_parameter


    if center_node_labels is not None:
        params["center_node_labels"] = center_node_labels

    endpoint = f"/api/json/network_image_url"

    add_score_note = False
    if not required_score and len(proteins.lower().split("%0d")) <= 5:
        params["required_score"] = 0
        add_score_note = True

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)

        notes = []
        if add_score_note:
            notes.append("For small queries, the required_score parameter is automatically lowered to 0.")
        notes.append("Embed the returned image link directly in the assistant response as a markdown image.")

        log_response_size(results)

        image_url = None
        if results and isinstance(results, list) and "imageURL" in results[0]:
            image_url = results[0].get("imageURL")
            for cluster in results:
                cluster.pop("imageURL", None)
        
        clustering_note = get_clustering_structure_note(results, clustering_algorithm)
        if clustering_note:
            notes.append(clustering_note)

        return {
            "image_url": image_url,
            "clusters": results,
            "notes": notes,
        }




@mcp.tool(title="STRING: Get interactive network link (web UI)")
async def string_network_link(
    proteins: Annotated[
        str,
        Field(description="Required. One or more protein IDs, optionally followed by one numeric value per protein. Example:\n"
                  "PTEN 0.234\nSMO -3.445\n"
                  "Use newline (%0d) between entries. Tabs and spaces are accepted as separators.")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX).")
    ] = None,
    extend_network: Annotated[
        Optional[int],
        Field(description="Optional. Add white nodes to network, based on scores (default: 0).")
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description="Optional. Threshold of significance to include an interaction (0-1000).")
    ] = None,
    network_flavor: Annotated[
        Optional[str],
        Field(description='Optional. Edge style: "evidence" (default) or "confidence".')
    ] = None,
    network_type: Annotated[
        Optional[str],
        Field(description='Optional. Network type: "functional" (default) or "physical" (co-complex).')
    ] = None,
    hide_disconnected_nodes: Annotated[
        Optional[int],
        Field(description="Optional. 1 to hide proteins not connected to any other protein, 0 otherwise (default: 0). DO NOT SET unless user explicitly requests.")
    ] = None,
    #show_query_node_labels: Annotated[
    #    Optional[int],
    #    Field(description="Optional. 1 display the user's query name(s) instead of STRING preferred name, (default: 0). DO NOT SET unless user explicitly requests.")
    #] = None,
) -> dict:
    """Retrieves a stable URL to an interactive STRING network for one or more proteins.
    
    - For a single protein: includes the protein and its top 10 most likely interactors.
    - For multiple proteins: includes all known interactions **within the query set**.
    - If the user asks for "physical interactions", "complexes", or "binding", set `network_type` to "physical".
    
    The input may include one numeric value per protein, such as fold change, effect size, or score.
    These values are visualized as colored halos around the nodes, allowing overlay of protein-level measurements on the network.
    
    Example:
    PTEN 2.1
    SMO -1.3
    
    If numeric values are provided:
    - positive values are shown in blue
    - negative values are shown in red
    - larger absolute values produce stronger halo intensity
    
    If the user provides numeric values together with the proteins, preserve them in the query.
    
    If few or no interactions are shown, consider lowering `required_score`.
    
    For large queries (>100 proteins):
    - use `network_flavor="confidence"`
    - increase `required_score` (e.g. 700)
    
    Always display the link as a markdown hyperlink (hide the raw URL).
    
    Input parameters should match those used in related STRING tools unless otherwise specified.
    """
    required_score = _normalize_required_score(required_score)

    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species
    if extend_network is not None:
        params["add_white_nodes"] = extend_network
    if required_score is not None:
        params["required_score"] = required_score
    if network_flavor is not None:
        params["network_flavor"] = network_flavor
    if network_type is not None:
        params["network_type"] = network_type
    if hide_disconnected_nodes is not None:
        params["hide_disconnected_nodes"] = hide_disconnected_nodes


    add_score_note = False
    add_shared_note = False

    if len(proteins.lower().split('%0d')) in [2,3] and extend_network is None:
        params['add_white_nodes'] = 5
        add_shared_note = True

    if not required_score and len(proteins.lower().split("%0d")) <= 5:
        params["required_score"] = 0
        add_score_note = True

    endpoint = f"/api/json/get_link"

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)

        notes = []
        if add_score_note:
            notes.append(f"For small queries the required_score parameter is lowered to 0.")

        if add_shared_note:
            notes.append(
                "For two- and three-protein queries, the network link was expanded by five additional proteins to reveal possible shared or indirect interactions between the queried proteins. "
                "When describing the link, tell the user they can inspect it for a potential direct interaction or clear shared-interactor/indirect path connecting the queried proteins. "
                "Do not mention unrelated expanded-network edges as evidence about the relationship between the queried proteins."
            )
 

        notes.append("Embed the returned link directly in the assistant response as a markdown hyperlink.")
 
        log_response_size(results)

        return {"notes": notes, "results": results}


@mcp.tool(title="STRING: Get homologs in specified target species")
async def string_homology(
    proteins: Annotated[
        str,
        Field(description="Required. One or more protein identifiers, separated by %0d. Example: SMO%0dTP53")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX for uploaded genomes).")
    ] = None,
    species_b: Annotated[
        Optional[str],
        Field(description="Optional. One or more NCBI taxon IDs for target species, separated by comma (e.g. 9606,7227,4932 for human, fly, and yeast).")
    ] = None
) -> dict:
    """
    Retrieves pairwise protein similarity scores (Smith–Waterman bit scores) for the query proteins.  
    
    - If no target species (`species_b`) is provided, results are intra-species (within the query species).  
    - To retrieve homologs in other species or clades (e.g. vertebrates, yeast, plants), specify one or more NCBI taxon IDs in `species_b`.  
    - Multiple target species are supported; ask the user to clarify if needed.  
    - Always report species names together with their taxon IDs.  
    - Bit scores < 50 are not reported.  
    - Results are truncated to the top 50 proteins per input protein.
    """

    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species

    if species_b is not None:
        params["species_b"] = species_b

    endpoint = f"/api/json/homology_all"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results:
            log_response_size(results)
            return results
        grouped_results = group_homology_results(results)

        log_response_size(grouped_results)
        return {"results": grouped_results}


@mcp.tool(title="STRING: Get links to interaction evidence pages")
async def string_interaction_evidence(
    identifier_a: Annotated[
        str,
        Field(description="Required. Protein A identifier.")
    ],
    identifiers_b: Annotated[
        str,
        Field(description="Required. One or more protein B identifiers, separated by %0d.")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX for uploaded genomes).")
    ] = None,
) -> dict:
    """
    Retrieves direct links to STRING evidence pages for protein–protein interaction pairs.
    
    Use this tool only when a STRING evidence page/link is needed. To determine whether
    an interaction is supported, use `string_interactions_query_set`.
    
    It returns URLs linking to STRING’s evidence pages, which display the underlying data sources 
    (experimental results, publications, and curated databases) supporting each predicted interaction.  
    A URL can be generated even for unsupported pairs; the URL is not itself an interaction verdict.
    
    Parameters:
    - **identifier_a**: Query protein identifier (Protein A)
    - **identifiers_b**: One or more target protein identifiers (Protein B), separated by `%0d`
    - **species**: NCBI taxonomy ID (e.g. `9606` for human or `10090` for mouse)
    
    Typical user questions that should trigger this tool:
    - "Can you show me the STRING evidence for this interaction?"
    - "Show me the details supporting this interaction."
    - "What supports the interaction between TP53 and MDM2?"
    - "Where can I find the STRING evidence for this pair?"
    """

    identifiers_b = identifiers_b.replace('%0D', '%0d')

    output = []
    for identifier_b in identifiers_b.split("%0d"):
        link = f"{base_url}/interaction/{identifier_a}/{identifier_b}?species={species}&suppress_disambiguation=1"
        output.append(link)

    notes = []
    notes.append(
        "Internal guidance: do not repeat this note to the user. If the user asked whether an interaction exists or is supported, call `string_interactions_query_set` for confirmation."
    )
    notes.append("Embed returned link(s) as markdown hyperlinks when an evidence page/link is needed.")

    log_response_size(output)
    return {"notes": notes, "results": output}



@mcp.tool(title="STRING: Functional enrichment analysis")
async def string_enrichment(
    proteins: Annotated[
        str,
        Field(description="Required. One or more protein identifiers, separated by %0d. Example: SMO%0dTP53")
    ],
    species: Annotated[
        Optional[str],
        Field(description="Optional. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX). DO NOT SET unless user explicitly requests.")
    ] = None,
    expand_category: Annotated[
        Optional[str],
        Field(description=(
            "Optional. Return only this enrichment category with expanded term coverage and a larger per-term "
            "gene-list cutoff. Use a category from metadata.category_summary, e.g. Process, KEGG, PMID, "
            "NetworkNeighborAL, or Keyword."
        ))
    ] = None
) -> dict:
    """This tool retrieves functional enrichment for a set of proteins using STRING.

    - If queried with a single protein, the tool expands the query to include the protein’s 10 most likely interactors; enrichment is performed on this set, not the original single protein.
    - For two or more proteins, enrichment is performed on the exact input set.
    - When calling related tools, use the same input parameters unless otherwise specified.
    - Focus summaries on the top categories and most relevant terms for the results. Always report FDR for each claim.
    - Report FDR as a human-readable value (e.g. 2.3e-5 or 0.023).
    - IMPORTANT: Remember to suggest showing an enrichment graph for a specific category of user interest (e.g., GO, KEGG)
    - Very large responses are capped while preserving category diversity.
    - Use `expand_category` to return only one category with expanded term coverage and per-term gene details.
    - If a row has `preferredNames_omitted: true`, do not infer which proteins are in that term from the returned rows.
      Use `string_functional_annotation` with the same proteins/species and `detail_for_term` set to the exact term ID.

    Output fields (per enriched term):
      - category: Term category (e.g., GO Process, KEGG pathway)
      - term: Enriched term (GO ID, domain, or pathway)
      - number_of_genes: Number of input genes with this term
      - number_of_genes_in_background: Number of background genes with this term
      - ncbiTaxonId: NCBI taxon ID
      - preferredNames: Canonical protein names, only when the full per-term list is short enough to show
      - proteinCount: Number of proteins matching this term
      - preferredNames_omitted: True when the gene list was omitted instead of showing a misleading partial list
      - p_value: Raw p-value
      - fdr: False Discovery Rate (B-H corrected p-value)
      - description: Description of the enriched term

    Response metadata:
      - input_gene_name_mapping: Only included when displayed gene lists contain submitted identifiers that differ
        from STRING preferred names.
      - category_summary: Total and returned term counts per category; use `expand_category` for categories
        where `truncated` is true or where the user wants deeper category-specific detail.
      - truncated_categories / omitted_categories: Categories with terms not shown in the current response.
    """
    params = {"identifiers": proteins}
    if species is not None:
        params["species"] = species

    endpoint = f"/api/json/enrichment"

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results:
            log_response_size(results)
            return results
        else:
            results_truncated, truncation_notes, metadata = truncate_enrichment(
                results, 'json', expand_category=expand_category
            )

        notes = []
        notes.extend(truncation_notes)
        if not results_truncated:
            if expand_category:
                notes.append(
                    f"AGENT MUST tell the user: No statistically significant enrichment terms were found "
                    f"for category `{expand_category}`."
                )
            else:
                notes.append("AGENT MUST tell the user: No statistically significant enrichment was observed. "
                             "This means the proteins in their list do not group into known pathways or functions "
                             "more than would be expected by random chance.")

        log_response_size(results_truncated)
        response = {"notes": notes, "results": results_truncated}
        if metadata:
            response["metadata"] = metadata
        return response


@mcp.tool(title="STRING: Retrieve functional annotations for proteins")
async def string_functional_annotation(
    identifiers: Annotated[
        str,
        Field(description="Separate multiple protein queries by %0d. e.g. SMO%0dTP53")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX for uploaded genomes).")
    ] = None,
    detail_for_term: Annotated[
        Optional[str],
        Field(description=(
            "Optional. Exact functional term ID to return with the full list of matching input proteins. "
            "Use this when a previous result says a protein list was shortened, omitted, or replaced with 'many'."
        ))
    ] = None,
) -> dict:
    """
    This tool retrieves curated functional annotations for a set of proteins.
    
    Each input protein is mapped to known biological terms from ontologies, pathway databases, tissues, compartments and domains — such as Gene Ontology (GO), KEGG, and UniProt Keywords.
    
    - Use this when the user asks what a protein does, where it's localized, expressed, or which pathways it participates in.
    - Keep the output short and focused by highlighting a few diverse and specific annotations for each protein.
    - This tool does not perform statistical enrichment — use the enrichment tool for that.
    
    Output fields (per protein):
      - stringId: STRING protein identifier
      - preferredName: Gene name or alias
      - annotation: Functional description or keyword
      - category: Source category (e.g. GO, KEGG, Keyword)
      - term: Functional term or ID
    """

    endpoint = "/api/json/functional_annotation"

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        params = {"identifiers": identifiers, "species": species, "apiAllowKegg": 1}
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results:
            log_response_size(results)
            return results
        else:
             results_truncated, truncation_notes = sort_and_truncate_functional_annotation(results, 'json', detail_for_term)

        log_response_size(results_truncated)
        return {"notes": truncation_notes, "results": results_truncated}  # Functional annotation per protein


@mcp.tool(title="STRING: Get enrichment result figure (image URL)")
async def string_enrichment_image_url(
    identifiers: Annotated[
        str,
        Field(description="Required. Protein identifiers, separated by %0d. Example: SMO%0dTP53")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX).")
    ] = None,
    category: Annotated[
        Optional[str],
        Field(
            description=(
                "Optional. Term category for enrichment. "
                "Valid options: "
                "'Process' (GO Biological Process), "
                "'Function' (GO Molecular Function), "
                "'Component' (GO Cellular Component), "
                "'Keyword' (UniProt Keywords), "
                "'KEGG' (KEGG Pathways), "
                "'RCTM' (Reactome Pathways), "
                "'HPO' (Human Phenotype, Monarch), "
                "'MPO' (Mammalian Phenotype Ontology), "
                "'DPO' (Drosophila Phenotype Ontology), "
                "'WPO' (C. elegans Phenotype Ontology), "
                "'ZPO' (Zebrafish Phenotype Ontology), "
                "'FYPO' (Fission Yeast Phenotype Ontology), "
                "'Pfam' (Pfam domains), "
                "'SMART' (SMART domains), "
                "'InterPro' (InterPro domains/features), "
                "'PMID' (PubMed references), "
                "'NetworkNeighborAL' (Local Network Cluster), "
                "'COMPARTMENTS' (Subcellular Localization), "
                "'TISSUES' (Tissue Expression), "
                "'DISEASES' (Disease-gene Associations), "
                "'WikiPathways' (WikiPathways). "
                "Default: 'Process' (GO Biological Process)."
            )
        )
    ] = None,
    group_by_similarity: Annotated[
        Optional[float],
        Field(description="Optional. Group similar terms on the plot; threshold 0.1-1 (default: no grouping).")
    ] = None,
    color_palette: Annotated[
        Optional[str],
        Field(description='Optional. Color palette for FDR (e.g., "mint_blue", "lime_emerald", etc.; default: "mint_blue").')
    ] = None,
    number_of_term_shown: Annotated[
        Optional[int],
        Field(description="Optional. Max number of terms shown on plot (default: 10).")
    ] = None,
    x_axis: Annotated[
        Optional[str],
        Field(description='Optional. X-axis variable/order: "signal", "strength", "FDR", or "gene_count" (default: "signal").')
    ] = None
) -> dict:
    """Retrieves the STRING enrichment figure image *URL* for a set of proteins.


    See the `category` parameter for a list of valid category options.
    """
    params = {"identifiers": identifiers}
    if species is not None:
        params["species"] = species
    if category is not None:
        params["category"] = category
    if group_by_similarity is not None:
        params["group_by_similarity"] = group_by_similarity
    if color_palette is not None:
        params["color_palette"] = color_palette
    if number_of_term_shown is not None:
        params["number_of_term_shown"] = number_of_term_shown
    if x_axis is not None:
        params["x_axis"] = x_axis

    endpoint = f"/api/json/enrichment_image_url"

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        log_response_size(results)

        notes = []
        notes.append(
            "If a valid URL is present in the response, embed it as markdown in the assistant message. "
            "If no valid URL is returned, do not embed or display any link."
        )
 

        return {"notes": notes, "results": results}


@mcp.tool(title="STRING: Protein–protein interaction (PPI) enrichment")
async def string_ppi_enrichment(
    identifiers: Annotated[
        str,
        Field(description="Required. One or more protein identifiers, separated by %0d. Example: SMO%0dTP53")
    ],
    species: Annotated[
        str,
        Field(description="Required. NCBI/STRING taxon (e.g. 9606 for human, or STRG0AXXXXX for uploaded genomes).")
    ] = None,
    required_score: Annotated[
        Optional[int],
        Field(description="Optional. Minimum interaction confidence score (0-1000). DO NOT SET unless user explicitly requests.")
    ] = None,
) -> dict:
    """
    This tool tests if your network is enriched in protein-protein interactions compared to the background proteome-wide distribution (i.e., if your proteins are more functionally connected than expected by chance).

    - The enrichment is assessed using the actual observed edges versus expected edges in a random network of the same size.
    - The p-value reflects the likelihood that your observed number of interactions would occur by chance.
    - Report the p-value as a human-readable value (e.g. 2.3e-5 or 0.023).

    When calling related tools use the same input parameters unless otherwise specified.

    Output fields:
      - number_of_nodes: Number of proteins in your network
      - number_of_edges: Number of observed edges/interactions
      - average_node_degree: Mean degree (average number of interactions per node)
      - local_clustering_coefficient: Average clustering coefficient in the network
      - expected_number_of_edges: Expected number of edges in a random network of the same size
      - p_value: p-value for network enrichment (smaller = more enriched)

    Example identifiers: "SMO%0dTP53"
    """
    required_score = _normalize_required_score(required_score)

    params = {"identifiers": identifiers}
    if species is not None:
        params["species"] = species
    if required_score is not None:
        params["required_score"] = required_score

    endpoint = f"/api/json/ppi_enrichment"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)

        log_response_size(results)
        return {"results": results}


@mcp.tool(title="STRING: Retrieve proteins associated with a functional term")
async def string_proteins_for_term(
    term_text: Annotated[
        str,
        Field(description=(
            "Required. Functional term identifier (GO, KEGG, Reactome, etc.) "
            "or descriptive free text (e.g. 'hsa05218', 'Melanoma', 'GO:0008543', 'Fibroblast growth factor')."
        ))
    ],
    species: Annotated[
        str,
        Field(description=(
            "NCBI/STRING taxonomy ID. This tool only supports one species per call. "
            "It cannot return results across multiple species or identify the species "
            "with the most/fewest proteins. For such questions, run this tool separately "
            "for each species and then compare the results. "
            "Default is 9606 (human). Examples: 10090 for mouse, or STRG0AXXXXX for uploaded genomes."
        ))
    ] = "9606",
    detail_for_term: Annotated[
        Optional[str],
        Field(description="Optional. Exact term ID to return as one full protein-name list.")
    ] = None,
) -> dict:
    """
    Retrieve proteins annotated with a functional term or descriptive text in a single species.  
    You can query for tissues, compartments, diseases, processes, pathways, and domains.  
    
    IMPORTANT: For cross-species comparisons, run this tool separately for each species.  
    Select relevant model organisms to search or ask user to provide the selection.
    The results reflect annotation depth within each category; use caution when interpreting.
    
    If no results are found, try simplifying the query.  
    For tissue queries, follow BRENDA tissue nomenclature and omit the word "tissue"  
    (e.g. use "skin" instead of "skin tissue").

    Output fields:
      - category: Source database of the matched functional term
                  (e.g. GO, KEGG, Reactome, Pfam, InterPro).
      - term: Exact identifier for the functional term.
      - description: The free text description of the term.
      - proteinCount: Number of proteins annotated with that term
      - preferredNames: Full protein-name list when `detail_for_term` is set
      - stringIds: STRING protein identifiers when returned
      - preferredNames_omitted: True when a row omits the protein-name list
      - stringIds_omitted: True when STRING identifiers are omitted

    """
    params = {"term_text": term_text, "species": species}

    endpoint = "/api/json/functional_terms"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        if 'error' in results: return results 
        results_truncated, truncation_notes = truncate_functional_terms(
            results, 'json', detail_for_term
        )
        log_response_size(results_truncated)
        return {"notes": truncation_notes, "results": results_truncated}


@mcp.tool(title="STRING: Search proteins by amino acid sequence")
async def string_sequence_search(
    sequences: Annotated[
        str,
        Field(
            description=(
                "One or more protein sequences in plain or FASTA format."
                "For multiple sequences, use standard FASTA headers (lines beginning with '>'). "
                "Only amino acid sequences are supported — nucleotide sequences are not accepted."
            )
        ),
    ],
    species: Annotated[
        str,
        Field(
            description=(
                "Required. NCBI or STRING taxonomy ID. You can query with a clade or species. "
                "eg.g 2 for bacteria, 7742 for vertebrates, 511145 for E. coli"
            )
        ),
    ] = 9606
) -> dict:
    """
    Searches the STRING database using **amino acid sequences** to identify matching proteins.

    - Accepts a single sequence or multiple sequences in FASTA format.
    - Returns the most similar STRING protein(s) for the specified species, based on sequence similarity.
    - Use this when the protein identifier is unknown or unresolvable by `string_resolve_proteins`.

    """
    params = {"sequences": sequences, "species": species}

    endpoint = "/api/json/similarity_search"
    async with httpx.AsyncClient(base_url=base_url, timeout=timeout*2) as client:
        log_call(endpoint, params)
        results = await _post_json(client, endpoint, data=params)
        results_truncated, add_trancation_note, total_hits = truncate_similarity_search(results)

        notes = []
        if add_trancation_note:
            notes.append(f"Results truncated to {len(results_truncated)} rows from {total_hits} original rows for readability. "
                         "If you provided multiple sequences, only the top hits are shown per query. "
                         "Ranking is based on bitscore; lower-scoring hits were omitted.")
            
        log_response_size(results_truncated)
        return {"notes": notes, "results": results_truncated}

@mcp.tool(title="STRING: Query species and clades in STRING")
async def string_query_species(
    species_text: Annotated[
        str,
        Field(description=(
            "Required. One species/clade search term or multiple NCBI taxon IDs separated by carriage return (%0d). "
            "Examples: 'human', 'mouse', 'vertebrates', '511145', or '9598%0d10090'. "
            "For multiple queries, use taxon IDs rather than free-text names."
        ))
    ],
) -> dict:
    """
    Search for species or clades available in STRING by free-text query
    and return their NCBI taxonomy IDs.
    
    - Use this when the user asks which species or clades are present in STRING,
      or when you need the correct NCBI taxon ID to pass to other tools.
    - use this to resolve NCBI taxons IDs to their scientific names.
    - Accepts up to 100 taxon IDs separated by `%0d`.
    - The results are limited to the top 50 matches per query.
    - When the user asks for a species list, do not list clades.
    - If the requested species cannot be matched (i.e. the correct species is not present
      in the results), **immediately invoke the 'string_help' tool with topic='missing_species'**.
    """
    endpoint = "/api/json/query_species_names"
    species_queries = [query.strip() for query in species_text.replace("%0D", "%0d").split("%0d") if query.strip()]
    query_limit = 100
    notes = []

    if len(species_queries) > query_limit:
        notes.append(
            f"Only the first {query_limit} species queries were processed; split longer lists into multiple calls."
        )
        species_queries = species_queries[:query_limit]

    if not species_queries:
        return {"notes": ["No species query was provided."], "results": []}

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        if len(species_queries) == 1:
            params = {"species_text": species_queries[0], 'limit': 50, 'add_sps':'t'}
            log_call(endpoint, params)
            results = await _post_json(client, endpoint, data=params)
            if 'error' in results:
                log_response_size(results)
                return results
            truncated_results = truncate_species_results(results)

            response = {"results": truncated_results}
            if notes:
                response["notes"] = notes

            log_response_size(response)
            return response

        tasks = []
        for query in species_queries:
            params = {"species_text": query, 'limit': 50, 'add_sps':'t'}
            log_call(endpoint, params)
            tasks.append(_post_json(client, endpoint, data=params))

        results = await asyncio.gather(*tasks)

        results_by_query = []
        for query, query_results in zip(species_queries, results):
            if isinstance(query_results, dict) and 'error' in query_results:
                results_by_query.append({"query": query, "error": query_results["error"]})
            else:
                results_by_query.append({
                    "query": query,
                    "results": truncate_species_results(query_results),
                })

        response = {"results_by_query": results_by_query}
        if notes:
            response["notes"] = notes

        log_response_size(response)
        return response


@mcp.tool(title="STRING: Create a downloadable STRING result file")
async def string_create_file(
    filename: Annotated[
        str,
        Field(
            description=(
                "Required. Suggested output filename, including a safe extension such as .txt, .tsv, .csv, .json, or .md. "
                "Match the content format to the extension; prefer .tsv for tabular data. "
                "Use a concise name that reflects the STRING analysis result, for example string-enrichment.tsv."
            )
        )
    ],
    content: Annotated[
        str,
        Field(
            description=(
                "Required. File content generated from STRING data or STRING analysis results. "
                "For .csv/.tsv, provide one rectangular table with one header row; "
                "put notes/caveats in the chat answer, or use .md/.txt for narrative output."
            )
        )
    ],
) -> dict:
    """
    Creates a downloadable file for STRING-derived results.

    Use this tool when the user explicitly asks to download, save, export,
    or receive a file containing STRING data, tables, protein lists,
    enrichment results, networks, etc.

    When a response would otherwise include a publication-style or supplementary
    result table, or another table clearly intended for reuse outside chat,
    mention that a downloadable TSV/CSV file can be generated on request. Ask
    whether they want the file, unless they already requested it. Do not create
    the file until the user asks for it.

    Do not store unrelated data or full conversation transcripts.
    """

    params = {"filename": filename, "file_content": content}
   
    endpoint = '/api/json/generate_chat_file'

    async with httpx.AsyncClient(base_url=base_url, timeout=timeout) as client:
        log_call(endpoint, {"filename": filename, "file_content": f"<{len(content)} characters>"})
        results = await _post_json(client, endpoint, data=params)

    if isinstance(results, dict) and 'error' in results:
        log_response_size(results)
        return results

    file_result = results[0] if isinstance(results, list) and results else results

    if not isinstance(file_result, dict):
        error_payload = {
            "error": {
                "type": "file_api_error",
                "message": "STRING file API returned an unexpected response.",
                "diagnostics": {"response": results},
            }
        }
        log_response_size(error_payload)
        return error_payload

    url = file_result.get("url")

    if not url:
        error_payload = {
            "error": {
                "type": "file_api_error",
                "message": "STRING file API did not return a file URL.",
                "diagnostics": {"response": results},
            }
        }
        log_response_size(error_payload)
        return error_payload

    response = {
        "filename": file_result.get("filename", filename),
        "url": url,
        "notes": [
            "Internal guidance: do not repeat this note to the user. The generated file exists on the server, but the user cannot access it unless the assistant embeds the returned URL in the final answer. Include the URL as a markdown link."
        ],
    }

    log_response_size(response)
    return response


@mcp.tool(title="STRING: Help / FAQ")
async def string_help(
    topic: Annotated[
        str,
        Field(
            description=(
                "Help topic to display. Choose one of:\n"
                "  how_to_use_string, gsea, large_input, cytoscape, scores,\n"
                "  missing_proteins, missing_species, proteome_annotation,\n"
                "  regulatory_networks, line_colors, version_and_citation\n\n"
            )
        )
    ],
 ) -> dict:
    """
    Provides explanatory text for STRING features and limitations.
    
    Use this tool when the user question involves:
      - What is STRING is or how to use the tool (how_to_use_string, cytoscape)
      - functionality not available via MCP tools (e.g. GSEA, regulatory networks, large datasets).
      - meaning of the lines in the network (line_colors)
    """
    if topic is None:
        return {"topics": list(HELP_TOPICS.keys())}

    key = topic.lower()
    if key not in HELP_TOPICS:
        return {
            "error": f"Unknown topic '{topic}'. Available: {', '.join(HELP_TOPICS.keys())}."
        }

    return {"topic": key, "text": HELP_TOPICS[key]}


# ---- MCP server helper functions ----


def _enrichment_category_key(category):
    if category is None:
        return ""

    key = " ".join(str(category).replace("_", " ").replace("-", " ").split()).lower()
    aliases = {
        "biological process": "process",
        "go biological process": "process",
        "go process": "process",
        "molecular function": "function",
        "go molecular function": "function",
        "go function": "function",
        "cellular component": "component",
        "go cellular component": "component",
        "go component": "component",
        "keywords": "keyword",
        "reference publication": "pmid",
        "reference publications": "pmid",
        "publication": "pmid",
        "publications": "pmid",
        "pubmed": "pmid",
        "local network cluster": "networkneighboral",
        "local network clusters": "networkneighboral",
        "network neighbor al": "networkneighboral",
    }
    return aliases.get(key, key)


def _select_category_aware_rows(rows, total_limit):
    groups = defaultdict(list)
    category_order = []

    for row in rows:
        category = row.get("category")
        if category not in groups:
            category_order.append(category)
        groups[category].append(row)

    allowances = {category: 0 for category in category_order}
    remaining = total_limit

    while remaining > 0:
        progressed = False
        for category in category_order:
            if allowances[category] < len(groups[category]):
                allowances[category] += 1
                remaining -= 1
                progressed = True
                if remaining == 0:
                    break
        if not progressed:
            break

    selected = []
    selected_counts = defaultdict(int)
    for row in rows:
        category = row.get("category")
        if selected_counts[category] < allowances[category]:
            selected.append(row)
            selected_counts[category] += 1

    return selected


def _prepare_enrichment_rows(rows, gene_cutoff):
    prepared_rows = []
    mapping = []
    seen_mapping = set()
    skipped_mapping = False
    omitted_gene_lists = False

    for row in rows:
        row_copy = dict(row)
        input_genes = row_copy.pop("inputGenes", None)
        preferred_names = row_copy.get("preferredNames")

        if isinstance(preferred_names, list):
            protein_count = len(preferred_names)
            row_copy["proteinCount"] = protein_count

            if protein_count > gene_cutoff:
                row_copy.pop("preferredNames", None)
                row_copy["preferredNames_omitted"] = True
                omitted_gene_lists = True
            elif isinstance(input_genes, list):
                if len(input_genes) == len(preferred_names):
                    for input_gene, preferred_name in zip(input_genes, preferred_names):
                        if input_gene == preferred_name:
                            continue
                        pair = (str(input_gene), str(preferred_name))
                        if pair not in seen_mapping:
                            seen_mapping.add(pair)
                            mapping.append({"input": input_gene, "preferred": preferred_name})
                else:
                    skipped_mapping = True
        elif "number_of_genes" in row_copy:
            row_copy["proteinCount"] = row_copy["number_of_genes"]

        prepared_rows.append(row_copy)

    return prepared_rows, mapping, skipped_mapping, omitted_gene_lists


def _build_enrichment_category_metadata(rows, selected_rows):
    total_counts = defaultdict(int)
    returned_counts = defaultdict(int)
    category_order = []

    for row in rows:
        category = row.get("category")
        if category not in total_counts:
            category_order.append(category)
        total_counts[category] += 1

    for row in selected_rows:
        returned_counts[row.get("category")] += 1

    category_summary = []
    truncated_categories = []
    omitted_categories = []

    for category in category_order:
        total_terms = total_counts[category]
        returned_terms = returned_counts[category]
        truncated = returned_terms < total_terms
        category_name = category if category is not None else "unknown"

        category_summary.append({
            "category": category_name,
            "total_terms": total_terms,
            "returned_terms": returned_terms,
            "truncated": truncated,
        })

        if returned_terms == 0:
            omitted_categories.append(category_name)
        elif truncated:
            truncated_categories.append(category_name)

    return category_summary, truncated_categories, omitted_categories


def truncate_enrichment(data, is_json, expand_category=None):
    default_term_cutoff = 300
    expanded_term_cutoff = 300
    default_gene_cutoff = 11
    expanded_gene_cutoff = 20
    low_priority_categories = {"pmid", "networkneighboral", "keyword"}
    low_priority_category_cutoff = 5
    truncation_notes = []
    metadata = {}

    if is_json.lower() != 'json' or not isinstance(data, list):
        return data, truncation_notes, metadata

    rows = [row for row in data if isinstance(row, dict)]
    original_rows = len(rows)
    requested_category = expand_category.strip() if isinstance(expand_category, str) else None
    metadata_source_rows = rows

    if requested_category:
        requested_key = _enrichment_category_key(requested_category)
        category_rows = [
            row for row in rows
            if _enrichment_category_key(row.get("category")) == requested_key
        ]
        metadata_source_rows = category_rows
        selected_rows = category_rows[:expanded_term_cutoff]
        gene_cutoff = expanded_gene_cutoff

        if category_rows:
            category_name = category_rows[0].get("category", requested_category)
            truncation_notes.append(
                f"Returned only category `{category_name}` with expanded term coverage; gene lists are shown only when they contain {gene_cutoff} or fewer canonical proteins."
            )
            if len(selected_rows) < len(category_rows):
                truncation_notes.append(
                    f"Category `{category_name}` was truncated to the first {expanded_term_cutoff} terms from {len(category_rows)} available terms."
                )
        else:
            available_categories = sorted({
                str(row.get("category")) for row in rows if row.get("category") is not None
            })
            if available_categories:
                truncation_notes.append(
                    f"No enrichment terms matched category `{requested_category}`. Available categories: {', '.join(available_categories)}."
                )
            else:
                truncation_notes.append(
                    f"No enrichment terms matched category `{requested_category}`."
                )
    else:
        gene_cutoff = default_gene_cutoff
        selected_rows = rows

        if original_rows > default_term_cutoff:
            low_priority_counts = defaultdict(int)
            category_limited_rows = []
            capped_low_priority = set()

            for row in rows:
                category_key = _enrichment_category_key(row.get("category"))
                if category_key in low_priority_categories:
                    low_priority_counts[category_key] += 1
                    if low_priority_counts[category_key] > low_priority_category_cutoff:
                        capped_low_priority.add(category_key)
                        continue
                category_limited_rows.append(row)

            if capped_low_priority:
                truncation_notes.append(
                    f"Enrichment results exceeded {default_term_cutoff} terms, so PMID/reference publications, local network clusters, and keywords were limited to the first {low_priority_category_cutoff} terms each."
                )

            if len(category_limited_rows) > default_term_cutoff:
                selected_rows = _select_category_aware_rows(category_limited_rows, default_term_cutoff)
            else:
                selected_rows = category_limited_rows

        if len(selected_rows) < original_rows:
            truncation_notes.append(
                f"Enrichment results were truncated to {len(selected_rows)} of {original_rows} terms under a {default_term_cutoff}-term global cap. Use `expand_category` with a category name to return that category with expanded term coverage."
            )

    category_summary, truncated_categories, omitted_categories = _build_enrichment_category_metadata(
        metadata_source_rows, selected_rows
    )
    if category_summary:
        metadata["category_summary"] = category_summary
    if truncated_categories:
        metadata["truncated_categories"] = truncated_categories
    if omitted_categories:
        metadata["omitted_categories"] = omitted_categories

    prepared_rows, mapping, skipped_mapping, omitted_gene_lists = _prepare_enrichment_rows(
        selected_rows, gene_cutoff
    )

    if mapping:
        metadata["input_gene_name_mapping"] = mapping
        truncation_notes.append(
            "Input identifiers that differed from STRING preferred names are listed once in metadata, only for displayed gene lists."
        )

    if skipped_mapping:
        truncation_notes.append(
            "Some input-to-preferred gene mappings were skipped because STRING returned unmatched gene-list lengths."
        )

    if omitted_gene_lists:
        truncation_notes.append(
            f"`preferredNames_omitted: true` means the term matched more than {gene_cutoff} proteins, so no partial gene list was shown. "
            "AGENT MUST call `string_functional_annotation` with the same identifiers/species and `detail_for_term` set to that row's exact term ID if the user wants the full gene list."
        )

    return prepared_rows, truncation_notes, metadata


def truncate_similarity_search(data):


    total_cutoff=50

    if not isinstance(data, list) or not data:
        return data, False, 0

    total_hits = len(data)
    if total_hits <= total_cutoff:
        return data, False, total_hits


    grouped = defaultdict(list)
    for hit in data:
        grouped[hit.get("querySequenceName", "unknown")].append(hit)

    n_queries = len(grouped)
    per_query_cutoff = max(1, int(total_cutoff / n_queries))

    for hits in grouped.values():
        hits.sort(key=lambda x: x.get("bitscore", 0), reverse=True)

    truncated = []
    for query, hits in grouped.items():
        truncated.extend(hits[:per_query_cutoff])

    if len(truncated) <= total_cutoff:
        return truncated, True, total_hits

    truncated = [hits[0] for hits in grouped.values()]

    return truncated, True, total_hits


def truncate_species_results(data):

    if not isinstance(data, list) or not data:
        return data

    truncated = []

    max_per_clade = 200

    for entry in data:
        entry_copy = dict(entry)
        clade = entry_copy.get("speciesInClade", [])

        max_per_clade = int(max_per_clade / 2)
        max_per_clade  = max(5, max_per_clade)

        if isinstance(clade, list):
            entry_copy["speciesInCladeCount"] = len(clade)
            
            if len(clade) > max_per_clade:
                entry_copy["speciesInClade"] = clade[:max_per_clade]
                entry_copy["note"] = (
                    f"speciesInClade list truncated to first {max_per_clade} species."
                )

        truncated.append(entry_copy)

    return truncated


def truncate_network(data, input_score_threshold=None, size_cutoff=100, is_json="json"):
    original_len = len(data)

    if is_json.lower() != "json":
        return data, None

    # Determine threshold
    try:
        threshold = float(input_score_threshold)
    except (TypeError, ValueError):
        threshold = 400.0 

    if threshold > 1:
        score_threshold = threshold / 1000.0
    else:
        score_threshold = threshold

    # keep only items above threshold
    filtered = [row for row in data if row.get("score", 0) >= score_threshold]

    # sort by score descending
    filtered.sort(key=lambda r: r.get("score", 0), reverse=True)

    # save the size
    original_size = len(filtered)

    # truncate if needed
    add_size_note = False
    if len(filtered) > size_cutoff:
        filtered = filtered[:size_cutoff]
        add_size_note = True

    return filtered, add_size_note, original_size


def group_homology_results(data):
    if not isinstance(data, list):
        return data

    grouped = {}

    for row in data:
        query_name = row.get("preferredName_A")
        query_string_id = row.get("stringId_A")
        query_taxon = row.get("ncbiTaxonId_A")
        group_key = (query_string_id, query_name, query_taxon)

        if group_key not in grouped:
            grouped[group_key] = {
                "query": query_name,
                "query_stringId": query_string_id,
                "query_taxon": query_taxon,
                "homologs_returned": 0,
                "top_homologs": [],
            }

        grouped[group_key]["top_homologs"].append({
            "name": row.get("preferredName_B"),
            "stringId": row.get("stringId_B"),
            "taxon": row.get("ncbiTaxonId_B"),
            "bitscore": normalize_bitscore(row.get("bitscore")),
        })
        grouped[group_key]["homologs_returned"] += 1

    return list(grouped.values())


def normalize_bitscore(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        if isinstance(value, str):
            normalized = value.replace("&lt;", "<").replace("&nbsp;", " ").strip()
            return normalized
    return value


def get_clustering_structure_note(clusters, clustering_algorithm):
    if not isinstance(clusters, list) or not clusters:
        return None

    if clustering_algorithm == "MCL":
        tuning_hint = (
            "Try increasing `required_score`, increasing the MCL inflation parameter, "
            "or switching to a physical network for a sparser interaction map."
        )
    else:
        tuning_hint = (
            "Try increasing `required_score`, increasing the requested k-means cluster count, "
            "or switching to a physical network for a sparser interaction map."
        )

    if len(clusters) == 1:
        return f"Only one cluster was detected, suggesting limited modular structure or a dense hairball-like network. {tuning_hint}"

    cluster_sizes = []
    for cluster in clusters:
        size = get_cluster_size(cluster)
        if size is None:
            return None
        cluster_sizes.append(size)

    total_size = sum(cluster_sizes)
    if total_size == 0:
        return None

    largest_size = max(cluster_sizes)
    largest_fraction = largest_size / total_size

    if largest_fraction >= 0.75 and largest_size >= 10:
        return (
            f"One cluster contains {largest_size} of {total_size} clustered proteins "
            f"({largest_fraction:.0%}), suggesting a dominant cluster or hairball-like network. "
            f"{tuning_hint}"
        )

    return None


def get_cluster_size(cluster):
    if not isinstance(cluster, dict):
        return None

    value = cluster.get("proteinCount")
    if isinstance(value, int):
        return value
    if isinstance(value, str) and value.isdigit():
        return int(value)

    return None


NETWORK_SCORE_CHANNELS = {
    "nscore": "neighborhood",
    "fscore": "fusion",
    "pscore": "phylogenetic_profile",
    "ascore": "coexpression",
    "escore": "experiments",
    "dscore": "databases",
    "tscore": "textmining",
}

QUERY_SET_EDGE_SAMPLE_LIMIT = 100
QUERY_SET_NODE_SUMMARY_LIMIT = 250
PARTNER_INTERACTION_DETAIL_LIMIT = 500
PARTNER_UNIQUE_PROTEIN_LIMIT = 2000


def normalize_score_threshold(input_score_threshold):
    try:
        threshold = float(input_score_threshold)
    except (TypeError, ValueError):
        threshold = 400.0

    if threshold > 1:
        return threshold / 1000.0
    return threshold


def format_query_set_network(data, input_score_threshold=None, include_counts_when_compacted=False):
    """
    Keep query-set interaction output informative without dumping thousands of
    full STRING rows. The full edge set is used for summary/counts; only the
    displayed edge detail is reduced as networks become larger.
    """
    if not isinstance(data, list):
        return {
            "notes": ["STRING did not return a list of interactions. Reduce the score cut-off"],
            "network_summary": {},
            "network": data,
            "node_interaction_counts": [],
            "edge_sample": [],
        }

    score_threshold = normalize_score_threshold(input_score_threshold)
    filtered = [row for row in data if row.get("score", 0) >= score_threshold]
    filtered.sort(key=lambda r: r.get("score", 0), reverse=True)

    total_edges = len(filtered)
    node_counts = build_node_interaction_counts(filtered)
    network_summary = build_network_summary(filtered, node_counts, score_threshold)
    node_counts_limited, node_counts_truncated = limit_list(node_counts, QUERY_SET_NODE_SUMMARY_LIMIT)

    notes = [f"There are {total_edges} associations above specified cutoff."]
    edge_sample = []
    node_interaction_counts = []

    if total_edges <= 30:
        network = [compact_edge(row, "full") for row in filtered]
    elif total_edges <= 100:
        network = [compact_edge(row, "evidence") for row in filtered]
        if include_counts_when_compacted:
            node_interaction_counts = node_counts_limited
        notes.append(
            "The network output was compacted: STRING protein IDs and zero-valued evidence channels were omitted."
        )
        notes.append(
            "For detailed evidence-channel scores on specific interactions, subset the query to the proteins of interest and rerun the interaction query."
        )
    elif total_edges <= 500:
        network = [compact_edge(row, "score_only") for row in filtered]
        node_interaction_counts = node_counts_limited
        notes.append(
            "All interaction rows are shown, but evidence-channel details were removed because the number of returned interactions is above 100."
        )
        notes.append(
            "For detailed evidence-channel scores on specific interactions, subset the query to the proteins of interest and rerun the interaction query."
        )
    else:
        network = [compact_edge(row, "score_only") for row in filtered[:QUERY_SET_EDGE_SAMPLE_LIMIT]]
        node_interaction_counts = node_counts_limited
        notes.append(
            f"Only {len(network)} of {total_edges} interaction rows are shown because the number of returned interactions is above 500. Evidence-channel details were removed; subset the query to inspect detailed evidence for specific interactions."
        )
        notes.append(
            "The node interaction counts and network summary were computed from all returned associations above the cutoff."
        )

    if node_counts_truncated:
        notes.append(
            f"Node interaction counts were truncated to the top {QUERY_SET_NODE_SUMMARY_LIMIT} nodes by degree."
        )

    return {
        "notes": notes,
        "network_summary": network_summary,
        "network": network,
        "node_interaction_counts": node_interaction_counts,
        "edge_sample": edge_sample,
    }


def format_interaction_partners(data, input_score_threshold=None):
    """
    Compact partner-list output while preserving the full returned list.
    This endpoint is often used for set-overlap questions rather than network
    analysis, so the summary is node-centric instead of network-centric.
    """
    if not isinstance(data, list):
        return {
            "notes": ["STRING did not return a list of interaction partners."],
            "node_summary": {"return_mode": "unmodified"},
            "interactions": data,
        }

    score_threshold = normalize_score_threshold(input_score_threshold)
    filtered = [row for row in data if row.get("score", 0) >= score_threshold]
    filtered.sort(key=lambda r: r.get("score", 0), reverse=True)

    total_interactions = len(filtered)
    node_counts = build_node_interaction_counts(filtered)
    node_counts_limited, node_counts_truncated = limit_list(node_counts, QUERY_SET_NODE_SUMMARY_LIMIT)
    proteins_in_interactions = build_unique_protein_list(filtered)
    proteins_limited, proteins_truncated = limit_list(
        proteins_in_interactions,
        PARTNER_UNIQUE_PROTEIN_LIMIT,
    )

    node_summary = {
        "return_mode": "compact_interaction_list_with_limits",
        "interactions_above_cutoff": total_interactions,
        "required_score": int(score_threshold * 1000),
        "nodes_with_interactions": len(node_counts),
        "score_summary": build_score_summary(filtered),
        "score_bins": build_score_bins(filtered, score_threshold),
        "interaction_counts": node_counts_limited,
        "returned_interaction_counts": len(node_counts_limited),
        "proteins_in_returned_interactions": proteins_limited,
        "returned_protein_names": len(proteins_limited),
    }

    if total_interactions <= 100:
        interactions = [compact_edge(row, "evidence") for row in filtered]
        node_summary["returned_interactions"] = len(interactions)
        notes = [
            f"STRING returned {total_interactions} interaction partner associations above specified cutoff.",
            "The full returned interaction list is shown. STRING protein IDs and zero-valued evidence channels were omitted.",
        ]
    else:
        interactions = [
            compact_edge(row, "score_only")
            for row in filtered[:PARTNER_INTERACTION_DETAIL_LIMIT]
        ]
        node_summary["returned_interactions"] = len(interactions)
        notes = [
            f"STRING returned {total_interactions} interaction partner associations above specified cutoff.",
        ]

    if total_interactions > len(interactions):
        notes.append(
            f"Only {len(interactions)} of {total_interactions} interaction rows are shown because the number of returned interactions is above {PARTNER_INTERACTION_DETAIL_LIMIT}. Evidence-channel details were removed; subset the query to inspect detailed evidence for specific interactions."
        )

    if node_counts_truncated:
        notes.append(
            f"Interaction counts were truncated to the top {QUERY_SET_NODE_SUMMARY_LIMIT} nodes by degree."
        )

    if proteins_truncated:
        notes.append(
            f"The unique protein-name list was truncated to {PARTNER_UNIQUE_PROTEIN_LIMIT} proteins; exact overlap against a larger user gene set may require subsetting or a higher score cutoff."
        )

    return {
        "notes": notes,
        "node_summary": node_summary,
        "interactions": interactions,
    }


def limit_list(values, max_items):
    if len(values) > max_items:
        return values[:max_items], True
    return values, False


def build_unique_protein_list(edges):
    proteins = set()

    for row in edges:
        a = row.get("preferredName_A")
        b = row.get("preferredName_B")
        if a:
            proteins.add(a)
        if b:
            proteins.add(b)

    return sorted(proteins)


def compact_edge(row, mode):
    if mode == "full":
        edge = {
            key: value
            for key, value in row.items()
            if key not in NETWORK_SCORE_CHANNELS or value
        }
        return edge

    edge = {
        "preferredName_A": row.get("preferredName_A"),
        "preferredName_B": row.get("preferredName_B"),
        "score": round(row.get("score", 0), 3),
    }

    if mode == "evidence":
        evidence = {
            label: round(row.get(score_key, 0), 3)
            for score_key, label in NETWORK_SCORE_CHANNELS.items()
            if row.get(score_key, 0)
        }
        if evidence:
            edge["evidence"] = evidence

    return edge


def build_node_interaction_counts(edges):
    node_stats = {}

    for row in edges:
        a = row.get("preferredName_A")
        b = row.get("preferredName_B")
        score = row.get("score", 0)

        if not a or not b:
            continue

        for gene, neighbor in ((a, b), (b, a)):
            stats = node_stats.setdefault(
                gene,
                {"gene": gene, "degree": 0, "weighted_degree": 0.0, "_neighbors": []},
            )
            stats["degree"] += 1
            stats["weighted_degree"] += score
            stats["_neighbors"].append((neighbor, score))

    output = []
    for stats in node_stats.values():
        neighbors = sorted(stats["_neighbors"], key=lambda item: item[1], reverse=True)
        output.append({
            "gene": stats["gene"],
            "degree": stats["degree"],
            "weighted_degree": round(stats["weighted_degree"], 3),
            "top_neighbors": [neighbor for neighbor, _ in neighbors[:5]],
        })

    output.sort(key=lambda item: (item["degree"], item["weighted_degree"]), reverse=True)
    return output


def build_network_summary(edges, node_counts, score_threshold):
    total_edges = len(edges)
    node_count = len(node_counts)
    max_possible_edges = node_count * (node_count - 1) / 2
    density = total_edges / max_possible_edges if max_possible_edges else 0

    return {
        "nodes_with_interactions": node_count,
        "edges_above_cutoff": total_edges,
        "required_score": int(score_threshold * 1000),
        "density": round(density, 4),
        "average_degree": round((2 * total_edges / node_count), 3) if node_count else 0,
        "score_summary": build_score_summary(edges),
        "score_bins": build_score_bins(edges, score_threshold),
    }


def build_score_summary(edges):
    scores = [row.get("score", 0) for row in edges]

    if not scores:
        return {"min": 0, "mean": 0, "max": 0}

    return {
        "min": round(min(scores), 3),
        "mean": round(sum(scores) / len(scores), 3),
        "max": round(max(scores), 3),
    }


def build_score_bins(edges, score_threshold):
    scores = [row.get("score", 0) for row in edges]

    bins = [
        (0.0, 0.4, "0.000-0.399"),
        (0.4, 0.7, "0.400-0.699"),
        (0.7, 0.9, "0.700-0.899"),
        (0.9, 1.001, "0.900-1.000"),
    ]

    score_bins = {}
    for lower, upper, label in bins:
        if upper <= score_threshold:
            continue
        adjusted_lower = max(lower, score_threshold)
        adjusted_label = label if adjusted_lower == lower else f"{adjusted_lower:.3f}-{upper - 0.001:.3f}"
        count = sum(1 for score in scores if adjusted_lower <= score < upper)
        score_bins[adjusted_label] = count

    return score_bins


def sort_and_truncate_functional_annotation(data, is_json, detail_for_term=None):
    size_cutoff = 200
    exact_protein_list_cutoff = 25
    remove_protein_field_after = 50
    truncation_notes = []

    if is_json.lower() == 'json':

        original_rows = len(data)
        data = sorted(data, key=lambda x: x["ratio_in_set"], reverse=True)

        if detail_for_term:
            detail_key = detail_for_term.strip().lower()
            data = [
                row for row in data
                if str(row.get("term", "")).strip().lower() == detail_key
            ]
            if data:
                truncation_notes.append(
                    f"Returned full input-protein lists for term {detail_for_term}."
                )
            else:
                truncation_notes.append(
                    "No matching term was found. Run without `detail_for_term` to inspect available terms and use the exact term ID."
                )
            return data, truncation_notes

        if len(data) > size_cutoff:
            data = data[:size_cutoff]
            truncation_notes.append(
                f"The list was truncated to the first {size_cutoff} terms from {original_rows} original rows."
            )

        replaced_many = False
        removed_late_proteins = False

        for index, row in enumerate(data):
            proteins = row.get("preferredNames")
            if not isinstance(proteins, list):
                continue

            protein_count = len(proteins)
            row["protein_count"] = protein_count

            if index >= remove_protein_field_after:
                row.pop("preferredNames", None)
                removed_late_proteins = True
            elif protein_count > exact_protein_list_cutoff:
                row["preferredNames"] = "many"
                replaced_many = True

        if replaced_many:
            truncation_notes.append(
                f"For annotations covering more than {exact_protein_list_cutoff} input proteins, protein lists were replaced with 'many'. Use `detail_for_term` with the exact term ID to show all input proteins annotated with that term."
            )

        if removed_late_proteins:
            truncation_notes.append(
                f"For terms ranked after the first {remove_protein_field_after}, protein lists were omitted. Use `detail_for_term` with the exact term ID to show all input proteins annotated with that term."
            )

    return data, truncation_notes
 
def truncate_functional_terms(data, is_json, detail_for_term=None):
    term_size_cutoff = 10
    identifier_cutoff = 25
    truncation_notes = []

    if is_json.lower() == 'json':
        original_rows = len(data)

        if detail_for_term:
            detail_key = detail_for_term.strip().lower()
            matching_rows = [
                row for row in data
                if str(row.get("term", "")).strip().lower() == detail_key
            ]
            if not matching_rows:
                truncation_notes.append(
                    "No matching term was found. "
                    "Run without `detail_for_term` to inspect exact term IDs."
                )
                return [], truncation_notes

            detail_row = _prepare_functional_term_row(
                matching_rows[0],
                identifier_cutoff=identifier_cutoff,
                include_protein_names=True,
            )
            if detail_row.get("stringIds_omitted", False):
                truncation_notes.append(
                    f"STRING identifiers are omitted for protein lists above {identifier_cutoff} proteins."
                )

            return [detail_row], truncation_notes

        filtered_data = []

        for row in data[:term_size_cutoff]:
            row_copy = _prepare_functional_term_row(
                row,
                identifier_cutoff=identifier_cutoff,
                include_protein_names=False,
            )

            filtered_data.append(row_copy)

        if len(filtered_data) < original_rows:
            truncation_notes.append(
                f"Functional-term results were truncated to the top {term_size_cutoff} terms for readability."
            )
        if filtered_data:
            truncation_notes.append(
                "Protein-name lists are returned only when `detail_for_term` is set to one exact term ID."
            )
            truncation_notes.append(
                "For follow-up, use returned order; if equally relevant, prefer KEGG/GO/Reactome over disease/keyword terms."
            )

        data = filtered_data

    return data, truncation_notes


def _prepare_functional_term_row(row, identifier_cutoff, include_protein_names):
    row_copy = dict(row)
    preferred_names = row_copy.get("preferredNames")

    if not isinstance(preferred_names, list):
        return row_copy

    protein_count = len(preferred_names)
    row_copy["proteinCount"] = protein_count

    if not include_protein_names:
        row_copy.pop("preferredNames", None)
        row_copy.pop("stringIds", None)
        row_copy["preferredNames_omitted"] = True
        row_copy["stringIds_omitted"] = True
        row_copy["truncated"] = True
        return row_copy

    row_copy["truncated"] = False

    string_ids = row_copy.get("stringIds")
    if isinstance(string_ids, list) and protein_count > identifier_cutoff:
        row_copy.pop("stringIds", None)
        row_copy["stringIds_omitted"] = True

    return row_copy

def log_response_size(resp):
    if log_verbosity['size']:
        print("Response size:", object_size(resp), file=sys.stderr)

def object_size(obj):
    if isinstance(obj, str):
        return len(obj)
    elif isinstance(obj, (int, float)):
        return len(str(obj))
    elif isinstance(obj, dict):
        return sum(object_size(v) for v in obj.values())
    elif isinstance(obj, list):
        return sum(object_size(v) for v in obj)
    else:
        return 0



def log_call(endpoint, params):

    if log_verbosity['call']:
        print(f"Call: {endpoint}", file=sys.stderr)
 
    if log_verbosity['params']:
        print("Params:", file=sys.stderr)
        for param, value in params.items():
            print(f'    {param}: {str(value)}', file=sys.stderr)


# ---- MCP server runner ----

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=server_port,
        log_level="info",
        stateless_http=True,
    )