# SPDX-License-Identifier: GPL-3.0-or-later
#
# Toolify: Empower any LLM with function calling capabilities.
# Copyright (C) 2025 FunnyCups (https://github.com/funnycups)
import os
import re
import json
import uuid
import asyncio
import httpx
import secrets
import string
import traceback
import time
import random
import logging
import tiktoken
import xml.etree.ElementTree as ET
from typing import List, Dict, Any, Optional, Literal, Union
from fastapi import FastAPI, Request, Header, HTTPException, Depends
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, ValidationError
from config_loader import config_loader
logger = logging.getLogger(__name__)
# Token Counter for counting tokens
class TokenCounter:
"""Token counter using tiktoken"""
# Model prefix to encoding mapping (from tiktoken source)
MODEL_PREFIX_TO_ENCODING = {
"o1-": "o200k_base",
"o3-": "o200k_base",
"o4-mini-": "o200k_base",
# chat
"gpt-5-": "o200k_base",
"gpt-4.5-": "o200k_base",
"gpt-4.1-": "o200k_base",
"chatgpt-4o-": "o200k_base",
"gpt-4o-": "o200k_base",
"gpt-4-": "cl100k_base",
"gpt-3.5-turbo-": "cl100k_base",
"gpt-35-turbo-": "cl100k_base", # Azure deployment name
"gpt-oss-": "o200k_harmony",
# fine-tuned
"ft:gpt-4o": "o200k_base",
"ft:gpt-4": "cl100k_base",
"ft:gpt-3.5-turbo": "cl100k_base",
"ft:davinci-002": "cl100k_base",
"ft:babbage-002": "cl100k_base",
}
def __init__(self):
self.encoders = {}
def get_encoder(self, model: str):
"""Get or create encoder for the model"""
if model not in self.encoders:
encoding = None
# First try to get encoding from model name directly
try:
self.encoders[model] = tiktoken.encoding_for_model(model)
return self.encoders[model]
except KeyError:
pass
# Try to find encoding by prefix matching
for prefix, enc_name in self.MODEL_PREFIX_TO_ENCODING.items():
if model.startswith(prefix):
encoding = enc_name
break
# Default to o200k_base for newer models
if encoding is None:
logger.warning(f"Model {model} not found in prefix mapping, using o200k_base encoding")
encoding = "o200k_base"
try:
self.encoders[model] = tiktoken.get_encoding(encoding)
except Exception as e:
logger.warning(f"Failed to get encoding {encoding} for model {model}: {e}. Falling back to cl100k_base")
self.encoders[model] = tiktoken.get_encoding("cl100k_base")
return self.encoders[model]
def count_tokens(self, messages: list, model: str = "gpt-3.5-turbo") -> int:
"""Count tokens in message list"""
encoder = self.get_encoder(model)
# All modern chat models use similar token counting
return self._count_chat_tokens(messages, encoder, model)
def _count_chat_tokens(self, messages: list, encoder, model: str) -> int:
"""Accurate token calculation for chat models
Based on OpenAI's token counting documentation:
- Each message has a fixed overhead
- Content tokens are counted per message
- Special tokens for message formatting
"""
# Token overhead varies by model
if model.startswith(("gpt-3.5-turbo", "gpt-35-turbo")):
# gpt-3.5-turbo uses different message overhead
tokens_per_message = 4 # <|start|>role<|separator|>content<|end|>
tokens_per_name = -1 # Name is omitted if not present
else:
# Most models including gpt-4, gpt-4o, o1, etc.
tokens_per_message = 3
tokens_per_name = 1
num_tokens = 0
for message in messages:
num_tokens += tokens_per_message
# Count tokens for each field in the message
for key, value in message.items():
if key == "content":
# Handle case where content might be a list (multimodal messages)
if isinstance(value, list):
for item in value:
if isinstance(item, dict) and item.get("type") == "text":
content_text = item.get("text", "")
num_tokens += len(encoder.encode(content_text, disallowed_special=()))
# Note: Image tokens are not counted here as they have fixed costs
elif isinstance(value, str):
num_tokens += len(encoder.encode(value, disallowed_special=()))
elif key == "name":
num_tokens += tokens_per_name
if isinstance(value, str):
num_tokens += len(encoder.encode(value, disallowed_special=()))
elif key == "role":
# Role is already counted in tokens_per_message
pass
elif isinstance(value, str):
# Other string fields
num_tokens += len(encoder.encode(value, disallowed_special=()))
# Every reply is primed with assistant role
num_tokens += 3
return num_tokens
def count_text_tokens(self, text: str, model: str = "gpt-3.5-turbo") -> int:
"""Count tokens in plain text"""
encoder = self.get_encoder(model)
return len(encoder.encode(text, disallowed_special=()))
# Global token counter instance
token_counter = TokenCounter()
def generate_random_trigger_signal() -> str:
"""Generate a random, self-closing trigger signal like ."""
chars = string.ascii_letters + string.digits
random_str = ''.join(secrets.choice(chars) for _ in range(4))
return f""
try:
app_config = config_loader.load_config()
log_level_str = app_config.features.log_level
if log_level_str == "DISABLED":
log_level = logging.CRITICAL + 1
else:
log_level = getattr(logging, log_level_str, logging.INFO)
logging.basicConfig(
level=log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
logger.info(f"✅ Configuration loaded successfully: {config_loader.config_path}")
logger.info(f"📊 Configured {len(app_config.upstream_services)} upstream services")
logger.info(f"🔑 Configured {len(app_config.client_authentication.allowed_keys)} client keys")
MODEL_TO_SERVICE_MAPPING, ALIAS_MAPPING = config_loader.get_model_to_service_mapping()
DEFAULT_SERVICE = config_loader.get_default_service()
ALLOWED_CLIENT_KEYS = config_loader.get_allowed_client_keys()
GLOBAL_TRIGGER_SIGNAL = generate_random_trigger_signal()
logger.info(f"🎯 Configured {len(MODEL_TO_SERVICE_MAPPING)} model mappings")
if ALIAS_MAPPING:
logger.info(f"🔄 Configured {len(ALIAS_MAPPING)} model aliases: {list(ALIAS_MAPPING.keys())}")
logger.info(f"🔄 Default service: {DEFAULT_SERVICE['name']}")
except Exception as e:
logger.error(f"❌ Configuration loading failed: {type(e).__name__}")
logger.error(f"❌ Error details: {str(e)}")
logger.error("💡 Please ensure config.yaml file exists and is properly formatted")
exit(1)
def build_tool_call_index_from_messages(messages: List[Dict[str, Any]]) -> Dict[str, Dict[str, str]]:
"""
Build tool_call_id -> {name, arguments} index from message history.
This replaces the server-side mapping by extracting tool calls from assistant messages.
Args:
messages: List of message dicts from the request
Returns:
Dict mapping tool_call_id to {name, arguments}
"""
index = {}
for msg in messages:
if isinstance(msg, dict) and msg.get("role") == "assistant":
tool_calls = msg.get("tool_calls")
if tool_calls and isinstance(tool_calls, list):
for tc in tool_calls:
if isinstance(tc, dict):
tc_id = tc.get("id")
func = tc.get("function", {})
if tc_id and isinstance(func, dict):
name = func.get("name", "")
arguments = func.get("arguments", "{}")
if not isinstance(arguments, str):
try:
arguments = json.dumps(arguments, ensure_ascii=False)
except Exception:
arguments = str(arguments)
if name:
index[tc_id] = {
"name": name,
"arguments": arguments
}
logger.debug(f"🔧 Indexed tool_call_id: {tc_id} -> {name}")
logger.debug(f"🔧 Built tool_call index with {len(index)} entries")
return index
def get_fc_error_retry_prompt(original_response: str, error_details: str) -> str:
custom_template = app_config.features.fc_error_retry_prompt_template
if custom_template:
return custom_template.format(
original_response=original_response,
error_details=error_details
)
return f"""Your previous response attempted to make a function call but the format was invalid or could not be parsed.
**Your original response:**
```
{original_response}
```
**Error details:**
{error_details}
**Instructions:**
Please retry and output the function call in the correct XML format. Remember:
1. Start with the trigger signal on its own line
2. Immediately follow with the XML block
3. Use with valid JSON for parameters
4. Do not add any text after
Please provide the corrected function call now. DO NOT OUTPUT ANYTHING ELSE."""
def _schema_type_name(v: Any) -> str:
if v is None:
return "null"
if isinstance(v, bool):
return "boolean"
if isinstance(v, int) and not isinstance(v, bool):
return "integer"
if isinstance(v, float):
return "number"
if isinstance(v, str):
return "string"
if isinstance(v, list):
return "array"
if isinstance(v, dict):
return "object"
return type(v).__name__
def _validate_value_against_schema(value: Any, schema: Dict[str, Any], path: str = "args", depth: int = 0) -> List[str]:
"""Best-effort JSON Schema validation for tool arguments.
Intentional subset:
- type, properties, required, additionalProperties
- items (array)
- enum, const
- anyOf/oneOf/allOf (basic)
- pattern/minLength/maxLength (string)
Returns a list of human-readable errors.
"""
if schema is None:
schema = {}
if depth > 8:
return [] # prevent pathological recursion
errors: List[str] = []
# Combinators
if isinstance(schema.get("allOf"), list):
for idx, sub in enumerate(schema["allOf"]):
errors.extend(_validate_value_against_schema(value, sub or {}, f"{path}.allOf[{idx}]", depth + 1))
return errors
if isinstance(schema.get("anyOf"), list):
option_errors = [
_validate_value_against_schema(value, sub or {}, path, depth + 1)
for sub in schema["anyOf"]
]
if not any(len(e) == 0 for e in option_errors):
errors.append(f"{path}: value does not satisfy anyOf options")
return errors
if isinstance(schema.get("oneOf"), list):
option_errors = [
_validate_value_against_schema(value, sub or {}, path, depth + 1)
for sub in schema["oneOf"]
]
ok_count = sum(1 for e in option_errors if len(e) == 0)
if ok_count != 1:
errors.append(f"{path}: value must satisfy exactly one oneOf option (matched {ok_count})")
return errors
# enum/const
if "const" in schema:
if value != schema.get("const"):
errors.append(f"{path}: expected const={schema.get('const')!r}, got {value!r}")
return errors
enum_vals = schema.get("enum")
if isinstance(enum_vals, list):
if value not in enum_vals:
errors.append(f"{path}: expected one of {enum_vals!r}, got {value!r}")
return errors
stype = schema.get("type")
if stype is None:
# If schema omits type but has object keywords, treat as object.
if any(k in schema for k in ("properties", "required", "additionalProperties")):
stype = "object"
# type checks
def _type_ok(t: str) -> bool:
if t == "object":
return isinstance(value, dict)
if t == "array":
return isinstance(value, list)
if t == "string":
return isinstance(value, str)
if t == "boolean":
return isinstance(value, bool)
if t == "integer":
return isinstance(value, int) and not isinstance(value, bool)
if t == "number":
return (isinstance(value, (int, float)) and not isinstance(value, bool))
if t == "null":
return value is None
return True
if isinstance(stype, str):
if not _type_ok(stype):
errors.append(f"{path}: expected type '{stype}', got '{_schema_type_name(value)}'")
return errors
elif isinstance(stype, list):
if not any(_type_ok(t) for t in stype if isinstance(t, str)):
errors.append(f"{path}: expected type in {stype!r}, got '{_schema_type_name(value)}'")
return errors
# string constraints
if isinstance(value, str):
min_len = schema.get("minLength")
max_len = schema.get("maxLength")
if isinstance(min_len, int) and len(value) < min_len:
errors.append(f"{path}: string shorter than minLength={min_len}")
if isinstance(max_len, int) and len(value) > max_len:
errors.append(f"{path}: string longer than maxLength={max_len}")
pat = schema.get("pattern")
if isinstance(pat, str):
try:
if re.search(pat, value) is None:
errors.append(f"{path}: string does not match pattern {pat!r}")
except re.error:
# ignore invalid patterns in schema
pass
# object
if isinstance(value, dict):
props = schema.get("properties")
if props is None:
props = {}
if not isinstance(props, dict):
props = {}
required = schema.get("required")
if required is None:
required = []
if not isinstance(required, list):
required = []
required = [k for k in required if isinstance(k, str)]
for k in required:
if k not in value:
errors.append(f"{path}: missing required property '{k}'")
additional = schema.get("additionalProperties", True)
for k, v in value.items():
if k in props:
errors.extend(_validate_value_against_schema(v, props.get(k) or {}, f"{path}.{k}", depth + 1))
else:
if additional is False:
errors.append(f"{path}: unexpected property '{k}'")
elif isinstance(additional, dict):
errors.extend(_validate_value_against_schema(v, additional, f"{path}.{k}", depth + 1))
# array
if isinstance(value, list):
items = schema.get("items")
if isinstance(items, dict):
for i, v in enumerate(value):
errors.extend(_validate_value_against_schema(v, items, f"{path}[{i}]", depth + 1))
return errors
def validate_parsed_tools(parsed_tools: List[Dict[str, Any]], tools: List["Tool"]) -> Optional[str]:
"""Validate parsed tool calls against declared tools definitions.
Returns a single error string if invalid, else None.
"""
tools = tools or []
allowed = {t.function.name: (t.function.parameters or {}) for t in tools if t and t.function and t.function.name}
allowed_names = sorted(list(allowed.keys()))
for idx, call in enumerate(parsed_tools or []):
name = (call or {}).get("name")
args = (call or {}).get("args")
if not isinstance(name, str) or not name:
return f"Tool call #{idx + 1}: missing tool name"
if name not in allowed:
return (
f"Tool call #{idx + 1}: unknown tool '{name}'. "
f"Allowed tools: {allowed_names}"
)
if not isinstance(args, dict):
return f"Tool call #{idx + 1} '{name}': arguments must be a JSON object, got {_schema_type_name(args)}"
schema = allowed[name] or {}
errs = _validate_value_against_schema(args, schema, path=f"{name}")
if errs:
# Keep message short but actionable
preview = "; ".join(errs[:6])
more = f" (+{len(errs) - 6} more)" if len(errs) > 6 else ""
return f"Tool call #{idx + 1} '{name}': schema validation failed: {preview}{more}"
return None
def _prompt_schema_type_name(schema: Any) -> str:
"""Return a compact type label for prompt display."""
if not isinstance(schema, dict):
return "any"
stype = schema.get("type")
if isinstance(stype, str):
return stype
if isinstance(stype, list):
parts = [t for t in stype if isinstance(t, str)]
return " | ".join(parts) if parts else "any"
if any(k in schema for k in ("properties", "required", "additionalProperties")):
return "object"
if "items" in schema:
return "array"
if isinstance(schema.get("anyOf"), list):
return "anyOf"
if isinstance(schema.get("oneOf"), list):
return "oneOf"
if isinstance(schema.get("allOf"), list):
return "allOf"
return "any"
def _prompt_schema_dump(value: Any) -> str:
try:
return json.dumps(value, ensure_ascii=False)
except Exception:
return str(value)
def _collect_prompt_schema_constraints(schema: Dict[str, Any]) -> Dict[str, Any]:
constraints: Dict[str, Any] = {}
for key in [
"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum",
"minLength", "maxLength", "pattern", "format",
"minItems", "maxItems", "uniqueItems",
"minProperties", "maxProperties", "multipleOf"
]:
if key in schema:
constraints[key] = schema.get(key)
if _prompt_schema_type_name(schema) == "array":
items = schema.get("items") or {}
if isinstance(items, dict):
item_type = _prompt_schema_type_name(items)
if item_type != "any":
constraints["items.type"] = item_type
return constraints
def _append_prompt_schema_body(
lines: List[str],
schema: Any,
is_required: Optional[bool],
indent_level: int,
depth: int = 0
) -> None:
schema_dict = schema if isinstance(schema, dict) else {}
indent = " " * indent_level
if depth > 8:
lines.append(f"{indent}- note: nested schema omitted after depth 8")
return
lines.append(f"{indent}- type: {_prompt_schema_type_name(schema_dict)}")
if is_required is not None:
lines.append(f"{indent}- required: {'Yes' if is_required else 'No'}")
description = schema_dict.get("description")
if description:
lines.append(f"{indent}- description: {description}")
enum_vals = schema_dict.get("enum")
if enum_vals is not None:
lines.append(f"{indent}- enum: {_prompt_schema_dump(enum_vals)}")
if "const" in schema_dict:
lines.append(f"{indent}- const: {_prompt_schema_dump(schema_dict.get('const'))}")
default_val = schema_dict.get("default")
if default_val is not None:
lines.append(f"{indent}- default: {_prompt_schema_dump(default_val)}")
examples_val = schema_dict.get("examples") or schema_dict.get("example")
if examples_val is not None:
lines.append(f"{indent}- examples: {_prompt_schema_dump(examples_val)}")
constraints = _collect_prompt_schema_constraints(schema_dict)
if constraints:
lines.append(f"{indent}- constraints: {_prompt_schema_dump(constraints)}")
props_raw = schema_dict.get("properties")
props = props_raw if isinstance(props_raw, dict) else {}
required_raw = schema_dict.get("required")
required_list = required_raw if isinstance(required_raw, list) else []
required_list = [k for k in required_list if isinstance(k, str)]
if required_list:
lines.append(f"{indent}- required properties: {', '.join(required_list)}")
if props:
lines.append(f"{indent}- properties:")
for child_name, child_schema in props.items():
child_indent = " " * (indent_level + 1)
child_name_text = str(child_name)
lines.append(f"{child_indent}- {child_name_text}:")
_append_prompt_schema_body(
lines,
child_schema,
child_name_text in required_list,
indent_level + 2,
depth + 1
)
items = schema_dict.get("items")
if isinstance(items, dict):
lines.append(f"{indent}- items:")
_append_prompt_schema_body(lines, items, None, indent_level + 1, depth + 1)
elif isinstance(items, list) and items:
lines.append(f"{indent}- items:")
for idx, item_schema in enumerate(items):
item_indent = " " * (indent_level + 1)
lines.append(f"{item_indent}- item[{idx}]:")
_append_prompt_schema_body(lines, item_schema, None, indent_level + 2, depth + 1)
additional = schema_dict.get("additionalProperties", True)
if additional is False:
lines.append(f"{indent}- additionalProperties: false")
elif isinstance(additional, dict):
lines.append(f"{indent}- additionalProperties:")
_append_prompt_schema_body(lines, additional, None, indent_level + 1, depth + 1)
for keyword in ("anyOf", "oneOf", "allOf"):
options = schema_dict.get(keyword)
if isinstance(options, list) and options:
lines.append(f"{indent}- {keyword}:")
for idx, option_schema in enumerate(options, start=1):
option_indent = " " * (indent_level + 1)
lines.append(f"{option_indent}- option {idx}:")
_append_prompt_schema_body(
lines,
option_schema,
None,
indent_level + 2,
depth + 1
)
async def attempt_fc_parse_with_retry(
content: str,
trigger_signal: str,
messages: List[Dict[str, Any]],
upstream_url: str,
headers: Dict[str, str],
model: str,
tools: List["Tool"],
timeout: int
) -> Optional[List[Dict[str, Any]]]:
"""
Attempt to parse function calls from content. If parsing fails and retry is enabled,
send error details back to the model for correction.
Returns parsed tool calls or None if parsing ultimately fails.
"""
def _parse_and_validate(current_content: str) -> tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
parsed = parse_function_calls_xml(current_content, trigger_signal)
if not parsed:
return None, None
validation_error = validate_parsed_tools(parsed, tools)
if validation_error:
return None, validation_error
return parsed, None
if not app_config.features.enable_fc_error_retry:
parsed, _err = _parse_and_validate(content)
return parsed
max_attempts = app_config.features.fc_error_retry_max_attempts
current_content = content
current_messages = messages.copy()
for attempt in range(max_attempts):
parsed_tools, validation_error = _parse_and_validate(current_content)
if parsed_tools:
if attempt > 0:
logger.info(f"✅ Function call parsing succeeded on retry attempt {attempt + 1}")
return parsed_tools
# IMPORTANT: only treat this as a tool-call attempt if the trigger signal appears
# outside of blocks. Otherwise models that mention the trigger
# inside think can spuriously trigger retries.
if find_last_trigger_signal_outside_think(current_content, trigger_signal) == -1:
logger.debug("🔧 No trigger signal found outside blocks; not a function call attempt")
return None
if attempt >= max_attempts - 1:
logger.warning(f"⚠️ Function call parsing failed after {max_attempts} attempts")
return None
# Classify the failure type to choose the right retry strategy
failure_type = _classify_fc_failure(current_content, trigger_signal)
if failure_type == "no_fc":
return None
error_details = validation_error or _diagnose_fc_parse_error(current_content, trigger_signal)
if failure_type == "truncated":
# Output was cut off — ask model to continue from where it stopped
retry_prompt = get_fc_continuation_prompt(current_content, error_details)
logger.info(f"🔄 Function call output truncated, requesting continuation {attempt + 2}/{max_attempts}")
else:
# Syntax error with closed tags — ask model to rewrite entirely
retry_prompt = get_fc_error_retry_prompt(current_content, error_details)
logger.info(f"🔄 Function call syntax error, requesting rewrite {attempt + 2}/{max_attempts}")
logger.debug(f"🔧 Failure type: {failure_type}, error details: {error_details}")
retry_messages = current_messages + [
{"role": "assistant", "content": current_content},
{"role": "user", "content": retry_prompt}
]
try:
retry_response = await http_client.post(
upstream_url,
json={"model": model, "messages": retry_messages, "stream": False},
headers=headers,
timeout=timeout
)
retry_response.raise_for_status()
retry_json = retry_response.json()
if retry_json.get("choices") and len(retry_json["choices"]) > 0:
retry_content = retry_json["choices"][0].get("message", {}).get("content", "")
logger.debug(f"🔧 Received retry response, length: {len(retry_content)}")
if failure_type == "truncated" and _is_continuation_response(retry_content, trigger_signal):
# Model chose Option A: continuation — merge with truncated content
current_content = _merge_truncated_and_continuation(current_content, retry_content)
logger.info(f"🔧 Merged continuation, total length: {len(current_content)}")
else:
# Model chose Option B (full rewrite) or it was a syntax_error retry
current_content = retry_content
current_messages = retry_messages
else:
logger.warning(f"⚠️ Retry response has no valid choices")
return None
except Exception as e:
logger.error(f"❌ Retry request failed: {e}")
return None
return None
def _diagnose_fc_parse_error(content: str, trigger_signal: str) -> str:
"""Diagnose why function call parsing failed and return error description."""
errors = []
if trigger_signal not in content:
errors.append(f"Trigger signal '{trigger_signal[:30]}...' not found in response")
return "; ".join(errors)
cleaned = remove_think_blocks(content)
if "" not in cleaned:
errors.append("Missing tag after trigger signal")
elif "" not in cleaned:
errors.append("Missing closing tag")
if "" not in cleaned:
errors.append("No blocks found inside ")
elif "" not in cleaned:
errors.append("Missing closing tag")
fc_match = re.search(r"([\s\S]*?)", cleaned)
if fc_match:
fc_content = fc_match.group(1)
if "" not in fc_content:
errors.append("Missing tag inside function_call")
if "" not in fc_content and "" not in fc_content:
errors.append("Missing or tag inside function_call")
args_json_match = re.search(r"([\s\S]*?)", fc_content)
if args_json_match:
args_content = args_json_match.group(1).strip()
cdata_match = re.search(r"", args_content)
json_to_parse = cdata_match.group(1) if cdata_match else args_content
try:
parsed = json.loads(json_to_parse)
if not isinstance(parsed, dict):
errors.append(f"args_json must be a JSON object, got {type(parsed).__name__}")
except json.JSONDecodeError as e:
errors.append(f"Invalid JSON in args_json: {str(e)}")
if not errors:
errors.append("XML structure appears correct but parsing failed for unknown reason")
return "; ".join(errors)
def _classify_fc_failure(content: str, trigger_signal: str) -> str:
"""Classify function call failure type.
Returns:
'no_fc' - no trigger signal found outside think blocks
'truncated' - trigger signal and opening tag found but no closing tag (output was cut off)
'syntax_error' - tags are closed but content is malformed
"""
if find_last_trigger_signal_outside_think(content, trigger_signal) == -1:
return "no_fc"
cleaned = remove_think_blocks(content)
pos = find_last_trigger_signal_outside_think(cleaned, trigger_signal)
if pos == -1:
return "no_fc"
after_trigger = cleaned[pos:]
has_open = "<" + "function_calls>" in after_trigger
has_close = "" in after_trigger
if not has_open:
return "syntax_error"
if has_open and not has_close:
return "truncated"
return "syntax_error"
def get_fc_continuation_prompt(truncated_content: str, error_details: str) -> str:
"""Generate a prompt asking the model to continue its truncated function call output."""
# Show only the tail to save tokens
tail = truncated_content[-1500:]
fci_close = ""
fc_close = ""
return (
"Your previous response was cut off before the function call XML was complete.\n"
"\n"
"**Your truncated response (ending abruptly):**\n"
"```\n"
f"{tail}\n"
"```\n"
"\n"
f"**What happened:** {error_details}\n"
"\n"
"**You have two options:**\n"
"\n"
"**Option A (PREFERRED \u2014 Continue writing):**\n"
"Output ONLY the exact continuation from where you were cut off. Rules:\n"
"- Start EXACTLY from the next character after the cutoff point \u2014 do not repeat ANY text, not even a single character\n"
"- If the cutoff happened mid-word, start from the next character of that word, never repeat the partial character/word\n"
"- Do NOT output any trigger signal or opening tags that were already present\n"
f"- End with the proper closing tags ({fci_close}, {fc_close} as needed)\n"
"- Do NOT add any explanation before or after\n"
"\n"
"**Option B (Only if you made an error earlier):**\n"
"Start fresh with the complete function call from the trigger signal. "
"Output the trigger signal on its own line, followed by the complete "
"function_calls block.\n"
"\n"
"Choose Option A unless you believe your previous output contained errors that need correction."
)
def _is_continuation_response(retry_content: str, trigger_signal: str) -> bool:
"""Determine if the model's retry response is a continuation or a full rewrite."""
cleaned = retry_content.strip()
if trigger_signal in cleaned:
return False
fc_open = "<" + "function_calls>"
if cleaned.lstrip().startswith(fc_open):
return False
return True
def _merge_truncated_and_continuation(truncated: str, continuation: str) -> str:
"""Merge truncated content with its continuation."""
return truncated.rstrip("\n") + continuation.lstrip("\n")
def format_tool_result_for_ai(tool_name: str, tool_arguments: str, result_content: str) -> str:
"""
Format tool call results for AI understanding with complete context.
Args:
tool_name: Name of the tool that was called
tool_arguments: Arguments passed to the tool (JSON string)
result_content: Execution result from the tool
Returns:
Formatted text for upstream model
"""
formatted_text = f"""Tool execution result:
- Tool name: {tool_name}
- Tool arguments: {tool_arguments}
- Execution result:
{result_content}
"""
logger.debug(f"🔧 Formatted tool result for {tool_name}")
return formatted_text
def format_assistant_tool_calls_for_ai(tool_calls: List[Dict[str, Any]], trigger_signal: str) -> str:
"""Format assistant tool calls into AI-readable string format."""
logger.debug(f"🔧 Formatting assistant tool calls. Count: {len(tool_calls)}")
def _wrap_cdata(text: str) -> str:
# Avoid illegal ']]>' sequence inside CDATA by splitting.
safe = (text or "").replace("]]>", "]]]]>")
return f""
xml_calls_parts = []
for tool_call in tool_calls:
function_info = tool_call.get("function", {})
name = function_info.get("name", "")
arguments_val = function_info.get("arguments", "{}")
# Strict: assistant.tool_calls must carry JSON-object arguments (or a JSON string representing an object).
try:
if isinstance(arguments_val, dict):
args_dict = arguments_val
elif isinstance(arguments_val, str):
parsed = json.loads(arguments_val or "{}")
if not isinstance(parsed, dict):
raise ValueError(f"arguments must be a JSON object, got {type(parsed).__name__}")
args_dict = parsed
else:
raise ValueError(f"arguments must be a JSON object or JSON string, got {type(arguments_val).__name__}")
except Exception as e:
raise HTTPException(
status_code=400,
detail=f"Invalid assistant.tool_calls arguments for tool '{name}': {e}"
)
args_payload = json.dumps(args_dict, ensure_ascii=False)
xml_call = (
f"\n"
f"{name}\n"
f"{_wrap_cdata(args_payload)}\n"
f""
)
xml_calls_parts.append(xml_call)
all_calls = "\n".join(xml_calls_parts)
final_str = f"{trigger_signal}\n\n{all_calls}\n"
logger.debug("🔧 Assistant tool calls formatted successfully.")
return final_str
def get_function_call_prompt_template(trigger_signal: str) -> str:
"""
Generate prompt template based on dynamic trigger signal
"""
custom_template = app_config.features.prompt_template
if custom_template:
logger.info("🔧 Using custom prompt template from configuration")
return custom_template.format(
trigger_signal=trigger_signal,
tools_list="{tools_list}"
)
return f"""
You have access to the following available tools to help solve problems:
{{tools_list}}
**IMPORTANT CONTEXT NOTES:**
1. You can call MULTIPLE tools in a single response if needed.
2. Even though you can call multiple tools, you MUST respect the user's later constraints and preferences (e.g., the user may request no tools, only one tool, or a specific tool/workflow).
3. The conversation context may already contain tool execution results from previous function calls. Review the conversation history carefully to avoid unnecessary duplicate tool calls.
4. When tool execution results are present in the context, they will be formatted with XML tags like ... for easy identification.
5. This is the ONLY format you can use for tool calls, and any deviation will result in failure.
When you need to use tools, you **MUST** strictly follow this format. Do NOT include any extra text, explanations, or dialogue on the first and second lines of the tool call syntax:
1. When starting tool calls, begin on a new line with exactly:
{trigger_signal}
No leading or trailing spaces, output exactly as shown above. The trigger signal MUST be on its own line and appear only once. Do not output a trigger signal for each tool call.
2. Starting from the second line, **immediately** follow with the complete XML block.
3. For multiple tool calls, include multiple blocks within the same wrapper, not separate blocks. Output the trigger signal only once, then one with all children.
4. Do not add any text or explanation after the closing tag.
STRICT ARGUMENT KEY RULES:
- You MUST use parameter keys EXACTLY as defined (case- and punctuation-sensitive). Do NOT rename, add, or remove characters.
- If a key starts with a hyphen (e.g., "-i", "-C"), you MUST keep the leading hyphen in the JSON key. Never convert "-i" to "i" or "-C" to "C".
- The tag must contain the exact name of a tool from the list. Any other tool name is invalid.
- The tag must contain a single JSON object with all required arguments for that tool.
- You MAY wrap the JSON content inside to avoid XML escaping issues.
CORRECT Example (multiple tool calls):
...response content (optional)...
{trigger_signal}
Grep
search
INCORRECT Example (extra text + wrong key names — DO NOT DO THIS):
...response content (optional)...
{trigger_signal}
I will call the tools for you.
Grep
true
2
.
INCORRECT Example (output non-XML format — DO NOT DO THIS):
...response content (optional)...
```json
{{"files":[{{"path":"system.py"}}]}}
```
Now please be ready to strictly follow the above specifications.
"""
class ToolFunction(BaseModel):
name: str
description: Optional[str] = None
parameters: Dict[str, Any]
class Tool(BaseModel):
type: Literal["function"]
function: ToolFunction
class Message(BaseModel):
role: str
content: Optional[str] = None
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
name: Optional[str] = None
class Config:
extra = "allow"
class ToolChoice(BaseModel):
type: Literal["function"]
function: Dict[str, str]
class ChatCompletionRequest(BaseModel):
model: str
messages: List[Dict[str, Any]]
tools: Optional[List[Tool]] = None
tool_choice: Optional[Union[str, ToolChoice]] = None
stream: Optional[bool] = False
stream_options: Optional[Dict[str, Any]] = None
temperature: Optional[float] = None
max_tokens: Optional[int] = None
top_p: Optional[float] = None
frequency_penalty: Optional[float] = None
presence_penalty: Optional[float] = None
n: Optional[int] = None
stop: Optional[Union[str, List[str]]] = None
class Config:
extra = "allow"
def generate_function_prompt(tools: List[Tool], trigger_signal: str) -> tuple[str, str]:
"""
Generate injected system prompt based on tools definition in client request.
Returns: (prompt_content, trigger_signal)
Raises:
HTTPException: If tool schema validation fails (e.g., required keys not in properties)
"""
tools_list_str = []
for i, tool in enumerate(tools):
func = tool.function
name = func.name
description = func.description or ""
# Robustly read JSON Schema fields + validate basic types
schema: Dict[str, Any] = func.parameters or {}
props_raw = schema.get("properties", {})
if props_raw is None:
props_raw = {}
if not isinstance(props_raw, dict):
raise HTTPException(
status_code=400,
detail=f"Tool '{name}': 'properties' must be an object, got {type(props_raw).__name__}"
)
props: Dict[str, Any] = props_raw
required_raw = schema.get("required", [])
if required_raw is None:
required_raw = []
if not isinstance(required_raw, list):
raise HTTPException(
status_code=400,
detail=f"Tool '{name}': 'required' must be a list, got {type(required_raw).__name__}"
)
non_string_required = [k for k in required_raw if not isinstance(k, str)]
if non_string_required:
raise HTTPException(
status_code=400,
detail=f"Tool '{name}': 'required' entries must be strings, got {non_string_required}"
)
required_list: List[str] = required_raw
missing_keys = [key for key in required_list if key not in props]
if missing_keys:
raise HTTPException(
status_code=400,
detail=f"Tool '{name}': required parameters {missing_keys} are not defined in properties"
)
# Brief summary line: name (type)
params_summary = ", ".join([
f"{p_name} ({_prompt_schema_type_name(p_info)})" for p_name, p_info in props.items()
]) or "None"
# Build detailed parameter spec for prompt injection (default enabled)
detail_lines: List[str] = []
for p_name, p_info in props.items():
detail_lines.append(f"- {p_name}:")
_append_prompt_schema_body(
detail_lines,
p_info,
p_name in required_list,
indent_level=1
)
detail_block = "\n".join(detail_lines) if detail_lines else "(no parameter details)"
desc_block = f"```\n{description}\n```" if description else "None"
tools_list_str.append(
f"{i + 1}. \n"
f" Description:\n{desc_block}\n"
f" Parameters summary: {params_summary}\n"
f" Required parameters: {', '.join(required_list) if required_list else 'None'}\n"
f" Parameter details:\n{detail_block}"
)
prompt_template = get_function_call_prompt_template(trigger_signal)
prompt_content = prompt_template.replace("{tools_list}", "\n\n".join(tools_list_str))
return prompt_content, trigger_signal
def remove_think_blocks(text: str) -> str:
"""
Temporarily remove all ... blocks for XML parsing
Supports nested think tags
Note: This function is only used for temporary parsing and does not affect the original content returned to the user
"""
while '' in text and '' in text:
start_pos = text.find('')
if start_pos == -1:
break
pos = start_pos + 7
depth = 1
while pos < len(text) and depth > 0:
if text[pos:pos+7] == '':
depth += 1
pos += 7
elif text[pos:pos+8] == '':
depth -= 1
pos += 8
else:
pos += 1
if depth == 0:
text = text[:start_pos] + text[pos:]
else:
break
return text
def find_last_trigger_signal_outside_think(text: str, trigger_signal: str) -> int:
"""
Find the last occurrence position of trigger_signal that is NOT inside any ... block.
Returns -1 if not found.
"""
if not text or not trigger_signal:
return -1
i = 0
think_depth = 0
last_pos = -1
while i < len(text):
if text.startswith("", i):
think_depth += 1
i += 7
continue
if text.startswith("", i):
think_depth = max(0, think_depth - 1)
i += 8
continue
if think_depth == 0 and text.startswith(trigger_signal, i):
last_pos = i
# Move forward by 1 to allow overlapping search (not expected, but safe)
i += 1
continue
i += 1
return last_pos
class StreamingFunctionCallDetector:
"""Enhanced streaming function call detector, supports dynamic trigger signals, avoids misjudgment within tags
Core features:
1. Avoid triggering tool call detection within blocks
2. Normally output block content to the user
3. Supports nested think tags
"""
def __init__(self, trigger_signal: str):
self.trigger_signal = trigger_signal
self.reset()
def reset(self):
self.content_buffer = ""
self.state = "detecting" # detecting, tool_parsing
self.in_think_block = False
self.think_depth = 0
self.signal = self.trigger_signal
self.signal_len = len(self.signal)
def process_chunk(self, delta_content: str) -> tuple[bool, str]:
"""
Process streaming content chunk
Returns: (is_tool_call_detected, content_to_yield)
"""
if not delta_content:
return False, ""
self.content_buffer += delta_content
content_to_yield = ""
if self.state == "tool_parsing":
return False, ""
if delta_content:
logger.debug(f"🔧 Processing chunk: {repr(delta_content[:50])}{'...' if len(delta_content) > 50 else ''}, buffer length: {len(self.content_buffer)}, think state: {self.in_think_block}")
i = 0
while i < len(self.content_buffer):
skip_chars = self._update_think_state(i)
if skip_chars > 0:
for j in range(skip_chars):
if i + j < len(self.content_buffer):
content_to_yield += self.content_buffer[i + j]
i += skip_chars
continue
if not self.in_think_block and self._can_detect_signal_at(i):
if self.content_buffer[i:i+self.signal_len] == self.signal:
logger.debug(f"🔧 Improved detector: detected trigger signal in non-think block! Signal: {self.signal[:20]}...")
logger.debug(f"🔧 Trigger signal position: {i}, think state: {self.in_think_block}, think depth: {self.think_depth}")
self.state = "tool_parsing"
self.content_buffer = self.content_buffer[i:]
return True, content_to_yield
remaining_len = len(self.content_buffer) - i
if remaining_len < self.signal_len or remaining_len < 8:
break
content_to_yield += self.content_buffer[i]
i += 1
self.content_buffer = self.content_buffer[i:]
return False, content_to_yield
def _update_think_state(self, pos: int):
"""Update think tag state, supports nesting"""
remaining = self.content_buffer[pos:]
if remaining.startswith(''):
self.think_depth += 1
self.in_think_block = True
logger.debug(f"🔧 Entering think block, depth: {self.think_depth}")
return 7
elif remaining.startswith(''):
self.think_depth = max(0, self.think_depth - 1)
self.in_think_block = self.think_depth > 0
logger.debug(f"🔧 Exiting think block, depth: {self.think_depth}")
return 8
return 0
def _can_detect_signal_at(self, pos: int) -> bool:
"""Check if signal can be detected at the specified position"""
return (pos + self.signal_len <= len(self.content_buffer) and
not self.in_think_block)
def finalize(self) -> Optional[List[Dict[str, Any]]]:
"""Final processing when stream ends"""
if self.state == "tool_parsing":
return parse_function_calls_xml(self.content_buffer, self.trigger_signal)
return None
def parse_function_calls_xml(xml_string: str, trigger_signal: str) -> Optional[List[Dict[str, Any]]]:
"""
Enhanced XML parsing function, supports dynamic trigger signals
1. Retain ... blocks (they should be returned normally to the user)
2. Temporarily remove think blocks only when parsing function_calls to prevent think content from interfering with XML parsing
3. Find the last occurrence of the trigger signal
4. Start parsing function_calls from the last trigger signal
"""
logger.debug(f"🔧 Improved parser starting processing, input length: {len(xml_string) if xml_string else 0}")
logger.debug(f"🔧 Using trigger signal: {trigger_signal[:20]}...")
if not xml_string or trigger_signal not in xml_string:
logger.debug(f"🔧 Input is empty or doesn't contain trigger signal")
return None
cleaned_content = remove_think_blocks(xml_string)
logger.debug(f"🔧 Content length after temporarily removing think blocks: {len(cleaned_content)}")
signal_positions = []
start_pos = 0
while True:
pos = cleaned_content.find(trigger_signal, start_pos)
if pos == -1:
break
signal_positions.append(pos)
start_pos = pos + 1
if not signal_positions:
logger.debug(f"🔧 No trigger signal found in cleaned content")
return None
logger.debug(f"🔧 Found {len(signal_positions)} trigger signal positions: {signal_positions}")
chosen_signal_index = None
chosen_signal_pos = None
calls_content_match = None
for idx in range(len(signal_positions) - 1, -1, -1):
pos = signal_positions[idx]
sub = cleaned_content[pos:]
m = re.search(r"([\s\S]*?)", sub)
if m:
chosen_signal_index = idx
chosen_signal_pos = pos
calls_content_match = m
logger.debug(f"🔧 Using trigger signal index {idx} at pos {pos}; content preview: {repr(sub[:100])}")
break
if calls_content_match is None:
logger.debug(f"🔧 No function_calls tag found after any trigger signal (triggers={len(signal_positions)})")
return None
calls_xml = calls_content_match.group(0)
calls_content = calls_content_match.group(1)
logger.debug(f"🔧 function_calls content: {repr(calls_content)}")
def _coerce_value(v: str):
try:
return json.loads(v)
except Exception:
return v
def _parse_args_json_payload(payload: str) -> Optional[Dict[str, Any]]:
"""Strict args_json parsing with markdown fence removal.
- Empty payload -> {}
- Strips markdown code fences (```json ... ```) if present
- Must be valid JSON and MUST decode to an object (dict)
- Any invalid / non-object payload -> None (treated as parse failure)
"""
if payload is None:
return {}
s = payload.strip()
if not s:
return {}
# Remove accidental markdown fences if model emits them
if s.startswith("```"):
s = re.sub(r"^```(?:json)?\s*", "", s)
s = re.sub(r"\s*```$", "", s)
try:
parsed = json.loads(s)
except Exception as e:
logger.debug(f"🔧 Invalid JSON in args_json: {type(e).__name__}: {e}")
return None
if not isinstance(parsed, dict):
logger.debug(f"🔧 args_json must decode to an object, got {type(parsed).__name__}")
return None
return parsed
def _extract_cdata_text(raw: str) -> str:
if raw is None:
return ""
if "", raw, flags=re.DOTALL)
return "".join(parts) if parts else raw
results: List[Dict[str, Any]] = []
# Primary path: strict XML parse (requires model to output valid XML)
try:
root = ET.fromstring(calls_xml)
for i, fc in enumerate(root.findall("function_call")):
tool_el = fc.find("tool")
name = (tool_el.text or "").strip() if tool_el is not None else ""
if not name:
logger.debug(f"🔧 No tool tag found in function_call #{i+1}")
continue
args: Dict[str, Any] = {}
args_json_el = fc.find("args_json")
if args_json_el is not None:
parsed_args = _parse_args_json_payload(args_json_el.text or "")
if parsed_args is None:
logger.debug(f"🔧 Invalid args_json in function_call #{i+1}; treating as parse failure")
return None
args = parsed_args
else:
# Legacy fallback: json
args_el = fc.find("args")
if args_el is not None:
for child in list(args_el):
args[child.tag] = _coerce_value(child.text or "")
result = {"name": name, "args": args}
results.append(result)
logger.debug(f"🔧 Added tool call: {result}")
logger.debug(f"🔧 Final parsing result (XML): {results}")
return results if results else None
except Exception as e:
logger.debug(f"🔧 XML library parse failed, falling back to regex parser: {type(e).__name__}: {e}")
# Fallback path: regex parse (more tolerant to malformed XML)
call_blocks = re.findall(r"([\s\S]*?)", calls_content)
logger.debug(f"🔧 Found {len(call_blocks)} function_call blocks")
for i, block in enumerate(call_blocks):
logger.debug(f"🔧 Processing function_call #{i+1}: {repr(block)}")
tool_match = re.search(r"(.*?)", block)
if not tool_match:
logger.debug(f"🔧 No tool tag found in block #{i+1}")
continue
name = tool_match.group(1).strip()
args: Dict[str, Any] = {}
args_json_match = re.search(r"([\s\S]*?)", block)
if args_json_match:
raw_payload = args_json_match.group(1)
payload = _extract_cdata_text(raw_payload)
parsed_args = _parse_args_json_payload(payload)
if parsed_args is None:
logger.debug(f"🔧 Invalid args_json in function_call #{i+1} (regex path); treating as parse failure")
return None
args = parsed_args
else:
# Legacy fallback
args_block_match = re.search(r"([\s\S]*?)", block)
if args_block_match:
args_content_inner = args_block_match.group(1)
arg_matches = re.findall(r"<([^\s>/]+)>([\s\S]*?)", args_content_inner)
for k, v in arg_matches:
args[k] = _coerce_value(v)
result = {"name": name, "args": args}
results.append(result)
logger.debug(f"🔧 Added tool call: {result}")
logger.debug(f"🔧 Final parsing result (regex): {results}")
return results if results else None
def find_upstream(model_name: str) -> tuple[Dict[str, Any], str]:
"""Find upstream configuration by model name, handling aliases and passthrough mode."""
# Handle model passthrough mode
if app_config.features.model_passthrough:
logger.info("🔄 Model passthrough mode is active. Forwarding to 'openai' service.")
openai_service = None
for service in app_config.upstream_services:
if service.name == "openai":
openai_service = service.model_dump()
break
if openai_service:
if not openai_service.get("api_key"):
raise HTTPException(status_code=500, detail="Configuration error: API key not found for the 'openai' service in model passthrough mode.")
# In passthrough mode, the model name from the request is used directly.
return openai_service, model_name
else:
raise HTTPException(status_code=500, detail="Configuration error: 'model_passthrough' is enabled, but no upstream service named 'openai' was found.")
# Default routing logic
chosen_model_entry = model_name
if model_name in ALIAS_MAPPING:
chosen_model_entry = random.choice(ALIAS_MAPPING[model_name])
logger.info(f"🔄 Model alias '{model_name}' detected. Randomly selected '{chosen_model_entry}' for this request.")
service = MODEL_TO_SERVICE_MAPPING.get(chosen_model_entry)
if service:
if not service.get("api_key"):
raise HTTPException(status_code=500, detail=f"Model configuration error: API key not found for service '{service.get('name')}'.")
else:
logger.warning(f"⚠️ Model '{model_name}' not found in configuration, using default service")
service = DEFAULT_SERVICE
if not service.get("api_key"):
raise HTTPException(status_code=500, detail="Service configuration error: Default API key not found.")
actual_model_name = chosen_model_entry
if ':' in chosen_model_entry:
parts = chosen_model_entry.split(':', 1)
if len(parts) == 2:
_, actual_model_name = parts
return service, actual_model_name
app = FastAPI()
http_client = httpx.AsyncClient()
def _is_retriable_upstream_error(exc: Exception) -> bool:
return isinstance(exc, (httpx.ConnectError, httpx.TimeoutException))
def _get_upstream_retry_attempts() -> int:
return getattr(app_config.server, "upstream_retry_attempts", 3) or 1
def _get_upstream_retry_delay(attempt: int) -> float:
base_delay = getattr(app_config.server, "upstream_retry_base_delay", 0.5)
return base_delay * (2 ** attempt)
async def _post_upstream_with_retry(url: str, json_body: dict, headers: Dict[str, str], timeout: int) -> httpx.Response:
"""POST to upstream with automatic retry on connection errors and timeouts."""
max_attempts = _get_upstream_retry_attempts()
if max_attempts <= 0:
return await http_client.post(url, json=json_body, headers=headers, timeout=timeout)
last_error: Optional[Exception] = None
for attempt in range(max_attempts):
try:
return await http_client.post(url, json=json_body, headers=headers, timeout=timeout)
except Exception as exc:
last_error = exc
if not _is_retriable_upstream_error(exc) or attempt >= max_attempts - 1:
raise
delay = _get_upstream_retry_delay(attempt)
logger.warning(
"⚠️ Upstream POST failed with %s; retrying in %.1fs (%s/%s)",
type(exc).__name__, delay, attempt + 2, max_attempts,
)
await asyncio.sleep(delay)
if last_error is not None:
raise last_error
raise RuntimeError("Unreachable upstream retry state")
@app.middleware("http")
async def debug_middleware(request: Request, call_next):
"""Middleware for debugging validation errors, does not log conversation content."""
response = await call_next(request)
if response.status_code == 422:
logger.debug(f"🔍 Validation error detected for {request.method} {request.url.path}")
logger.debug(f"🔍 Response status code: 422 (Pydantic validation failure)")
return response
@app.exception_handler(ValidationError)
async def validation_exception_handler(request: Request, exc: ValidationError):
"""Handle Pydantic validation errors with detailed error information"""
logger.error(f"❌ Pydantic validation error: {exc}")
logger.error(f"❌ Request URL: {request.url}")
logger.error(f"❌ Error details: {exc.errors()}")
for error in exc.errors():
logger.error(f"❌ Validation error location: {error.get('loc')}")
logger.error(f"❌ Validation error message: {error.get('msg')}")
logger.error(f"❌ Validation error type: {error.get('type')}")
return JSONResponse(
status_code=422,
content={
"error": {
"message": "Invalid request format",
"type": "invalid_request_error",
"code": "invalid_request"
}
}
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""Handle FastAPI HTTPException with OpenAI-compatible error envelope"""
status = exc.status_code
if status == 400:
err_type = "invalid_request_error"
code = "invalid_request"
elif status == 401:
err_type = "authentication_error"
code = "unauthorized"
elif status == 403:
err_type = "permission_error"
code = "forbidden"
elif status == 429:
err_type = "rate_limit_error"
code = "rate_limit_exceeded"
else:
err_type = "server_error"
code = "internal_error"
return JSONResponse(
status_code=status,
content={
"error": {
"message": str(exc.detail),
"type": err_type,
"code": code,
}
},
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""Handle all uncaught exceptions"""
logger.error(f"❌ Unhandled exception: {exc}")
logger.error(f"❌ Request URL: {request.url}")
logger.error(f"❌ Exception type: {type(exc).__name__}")
logger.error(f"❌ Error stack: {traceback.format_exc()}")
return JSONResponse(
status_code=500,
content={
"error": {
"message": "Internal server error",
"type": "server_error",
"code": "internal_error"
}
}
)
async def verify_api_key(authorization: str = Header(...)):
"""Dependency: verify client API key"""
client_key = authorization.replace("Bearer ", "")
if app_config.features.key_passthrough:
# In passthrough mode, skip allowed_keys check
return client_key
if client_key not in ALLOWED_CLIENT_KEYS:
raise HTTPException(status_code=401, detail="Unauthorized")
return client_key
def preprocess_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Preprocess messages:
- Convert role=tool messages to role=user text for upstream compatibility
- Convert assistant.tool_calls into assistant.content (XML format) for upstream context
- Convert developer->system if configured
"""
tool_call_index = build_tool_call_index_from_messages(messages)
processed_messages: List[Dict[str, Any]] = []
for message in messages:
if isinstance(message, dict):
if message.get("role") == "tool":
tool_call_id = message.get("tool_call_id")
content = message.get("content")
if not tool_call_id:
raise HTTPException(status_code=400, detail="Tool message missing tool_call_id")
# content may be empty string in some cases; only reject None
if content is None:
raise HTTPException(status_code=400, detail=f"Tool message missing content for tool_call_id={tool_call_id}")
tool_info = tool_call_index.get(tool_call_id)
if not tool_info:
raise HTTPException(
status_code=400,
detail=(
f"tool_call_id={tool_call_id} not found in conversation history. "
f"Ensure the assistant message with this tool_call is included in the messages array."
)
)
formatted_content = format_tool_result_for_ai(
tool_name=tool_info["name"],
tool_arguments=tool_info["arguments"],
result_content=content,
)
processed_messages.append({
"role": "user",
"content": formatted_content
})
logger.debug(f"🔧 Converted tool message to user message: tool_call_id={tool_call_id}, tool={tool_info['name']}")
elif message.get("role") == "assistant" and message.get("tool_calls"):
tool_calls = message.get("tool_calls", [])
formatted_tool_calls_str = format_assistant_tool_calls_for_ai(tool_calls, GLOBAL_TRIGGER_SIGNAL)
original_content = message.get("content") or ""
final_content = f"{original_content}\n{formatted_tool_calls_str}".strip()
processed_message = {
"role": "assistant",
"content": final_content
}
for key, value in message.items():
if key not in ["role", "content", "tool_calls"]:
processed_message[key] = value
processed_messages.append(processed_message)
logger.debug("🔧 Converted assistant tool_calls to content.")
elif message.get("role") == "developer":
if app_config.features.convert_developer_to_system:
processed_message = message.copy()
processed_message["role"] = "system"
processed_messages.append(processed_message)
logger.debug("🔧 Converted developer message to system message for better upstream compatibility")
else:
processed_messages.append(message)
logger.debug("🔧 Keeping developer role unchanged (based on configuration)")
else:
processed_messages.append(message)
else:
processed_messages.append(message)
return processed_messages
@app.post("/v1/chat/completions")
async def chat_completions(
request: Request,
body: ChatCompletionRequest,
_api_key: str = Depends(verify_api_key)
):
"""Main chat completion endpoint, proxy and inject function calling capabilities."""
start_time = time.time()
try:
logger.debug(f"🔧 Received request, model: {body.model}")
logger.debug(f"🔧 Number of messages: {len(body.messages)}")
logger.debug(f"🔧 Number of tools: {len(body.tools) if body.tools else 0}")
logger.debug(f"🔧 Streaming: {body.stream}")
upstream, actual_model = find_upstream(body.model)
upstream_url = f"{upstream['base_url']}/chat/completions"
logger.debug(f"🔧 Starting message preprocessing, original message count: {len(body.messages)}")
processed_messages = preprocess_messages(body.messages)
logger.debug(f"🔧 Preprocessing completed, processed message count: {len(processed_messages)}")
if not validate_message_structure(processed_messages):
logger.error(f"❌ Message structure validation failed, but continuing processing")
request_body_dict = body.model_dump(exclude_unset=True)
request_body_dict["model"] = actual_model
request_body_dict["messages"] = processed_messages
is_fc_enabled = app_config.features.enable_function_calling
has_tools_in_request = bool(body.tools)
has_function_call = is_fc_enabled and has_tools_in_request
logger.debug(f"🔧 Request body constructed, message count: {len(processed_messages)}")
except HTTPException as e:
# Preserve expected status codes (e.g., 400 for invalid tool_call_id history)
logger.error(f"❌ Request rejected: status_code={e.status_code}, detail={e.detail}")
return JSONResponse(
status_code=e.status_code,
content={
"error": {
"message": str(e.detail),
"type": "invalid_request_error" if e.status_code == 400 else (
"authentication_error" if e.status_code == 401 else (
"permission_error" if e.status_code == 403 else (
"rate_limit_error" if e.status_code == 429 else "server_error"
)
)
),
"code": "invalid_request" if e.status_code == 400 else (
"unauthorized" if e.status_code == 401 else (
"forbidden" if e.status_code == 403 else (
"rate_limit_exceeded" if e.status_code == 429 else "internal_error"
)
)
)
}
}
)
except Exception as e:
logger.error(f"❌ Request preprocessing failed: {str(e)}")
logger.error(f"❌ Error type: {type(e).__name__}")
if hasattr(app_config, 'debug') and app_config.debug:
logger.error(f"❌ Error stack: {traceback.format_exc()}")
return JSONResponse(
status_code=422,
content={
"error": {
"message": "Invalid request format",
"type": "invalid_request_error",
"code": "invalid_request"
}
}
)
if has_function_call:
logger.debug(f"🔧 Using global trigger signal for this request: {GLOBAL_TRIGGER_SIGNAL}")
tools_for_request: List[Tool] = body.tools or []
function_prompt, _ = generate_function_prompt(tools_for_request, GLOBAL_TRIGGER_SIGNAL)
tool_choice_prompt = safe_process_tool_choice(body.tool_choice, tools_for_request)
if tool_choice_prompt:
function_prompt += tool_choice_prompt
system_message = {"role": "system", "content": function_prompt}
request_body_dict["messages"].insert(0, system_message)
if "tools" in request_body_dict:
del request_body_dict["tools"]
if "tool_choice" in request_body_dict:
del request_body_dict["tool_choice"]
elif has_tools_in_request and not is_fc_enabled:
logger.info(f"🔧 Function calling is disabled by configuration, ignoring 'tools' and 'tool_choice' in request.")
if "tools" in request_body_dict:
del request_body_dict["tools"]
if "tool_choice" in request_body_dict:
del request_body_dict["tool_choice"]
prompt_tokens = token_counter.count_tokens(request_body_dict["messages"], body.model)
logger.info(f"📊 Request to {body.model} - Actual input tokens (including all preprocessing & injected prompts): {prompt_tokens}")
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {_api_key}" if app_config.features.key_passthrough else f"Bearer {upstream['api_key']}",
"Accept": "application/json" if not body.stream else "text/event-stream"
}
logger.info(f"📝 Forwarding request to upstream: {upstream['name']}")
logger.info(f"📝 Model: {request_body_dict.get('model', 'unknown')}, Messages: {len(request_body_dict.get('messages', []))}")
if not body.stream:
try:
logger.debug(f"🔧 Sending upstream request to: {upstream_url}")
logger.debug(f"🔧 has_function_call: {has_function_call}")
logger.debug(f"🔧 Request body contains tools: {bool(body.tools)}")
upstream_response = await _post_upstream_with_retry(
upstream_url, request_body_dict, headers, app_config.server.timeout
)
upstream_response.raise_for_status()
response_json: Dict[str, Any] = upstream_response.json()
logger.debug(f"🔧 Upstream response status code: {upstream_response.status_code}")
# Count output tokens and handle usage
completion_text = ""
if response_json.get("choices") and len(response_json["choices"]) > 0:
message = response_json["choices"][0].get("message", {})
# Extract content
content = message.get("content")
if content:
completion_text = content
# Check for reasoning_content
reasoning_content = message.get("reasoning_content")
if reasoning_content:
completion_text = (completion_text + "\n" + reasoning_content).strip() if completion_text else reasoning_content
logger.debug(f"🔧 Found reasoning_content, adding {len(reasoning_content)} chars to token count")
# Calculate our estimated tokens
estimated_completion_tokens = token_counter.count_text_tokens(completion_text, body.model) if completion_text else 0
estimated_prompt_tokens = prompt_tokens
estimated_total_tokens = estimated_prompt_tokens + estimated_completion_tokens
elapsed_time = time.time() - start_time
# Check if upstream provided usage and respect it
upstream_usage = response_json.get("usage", {})
if upstream_usage:
# Preserve upstream's usage structure and only replace zero values
final_usage = upstream_usage.copy()
# Replace zero or missing values with our estimates
if not final_usage.get("prompt_tokens") or final_usage.get("prompt_tokens") == 0:
final_usage["prompt_tokens"] = estimated_prompt_tokens
logger.debug(f"🔧 Replaced zero/missing prompt_tokens with estimate: {estimated_prompt_tokens}")
if not final_usage.get("completion_tokens") or final_usage.get("completion_tokens") == 0:
final_usage["completion_tokens"] = estimated_completion_tokens
logger.debug(f"🔧 Replaced zero/missing completion_tokens with estimate: {estimated_completion_tokens}")
if not final_usage.get("total_tokens") or final_usage.get("total_tokens") == 0:
final_usage["total_tokens"] = final_usage.get("prompt_tokens", estimated_prompt_tokens) + final_usage.get("completion_tokens", estimated_completion_tokens)
logger.debug(f"🔧 Replaced zero/missing total_tokens with calculated value: {final_usage['total_tokens']}")
response_json["usage"] = final_usage
logger.debug(f"🔧 Preserved upstream usage with replacements: {final_usage}")
else:
# No upstream usage, provide our estimates
response_json["usage"] = {
"prompt_tokens": estimated_prompt_tokens,
"completion_tokens": estimated_completion_tokens,
"total_tokens": estimated_total_tokens
}
logger.debug(f"🔧 No upstream usage found, using estimates")
# Log token statistics
actual_usage = response_json["usage"]
logger.info("=" * 60)
logger.info(f"📊 Token Usage Statistics - Model: {body.model}")
logger.info(f" Input Tokens: {actual_usage.get('prompt_tokens', 0)}")
logger.info(f" Output Tokens: {actual_usage.get('completion_tokens', 0)}")
logger.info(f" Total Tokens: {actual_usage.get('total_tokens', 0)}")
logger.info(f" Duration: {elapsed_time:.2f}s")
logger.info("=" * 60)
if has_function_call:
content = response_json["choices"][0]["message"]["content"]
logger.debug(f"🔧 Complete response content: {repr(content)}")
parsed_tools = await attempt_fc_parse_with_retry(
content=content,
trigger_signal=GLOBAL_TRIGGER_SIGNAL,
messages=request_body_dict["messages"],
upstream_url=upstream_url,
headers=headers,
model=actual_model,
tools=body.tools or [],
timeout=app_config.server.timeout
)
logger.debug(f"🔧 XML parsing result: {parsed_tools}")
if parsed_tools:
logger.debug(f"🔧 Successfully parsed {len(parsed_tools)} tool calls")
estimated_completion_tokens = token_counter.count_text_tokens(content, body.model)
estimated_total_tokens = estimated_prompt_tokens + estimated_completion_tokens
logger.debug(f"🔧 Completion tokens: {estimated_completion_tokens}")
tool_calls = []
for tool in parsed_tools:
tool_call_id = f"call_{uuid.uuid4().hex}"
tool_calls.append({
"id": tool_call_id,
"type": "function",
"function": {
"name": tool["name"],
"arguments": json.dumps(tool["args"])
}
})
logger.debug(f"🔧 Converted tool_calls: {tool_calls}")
prefix_pos = find_last_trigger_signal_outside_think(content, GLOBAL_TRIGGER_SIGNAL)
prefix_text = None
if prefix_pos != -1:
prefix_text = content[:prefix_pos].rstrip()
if prefix_text == "":
prefix_text = None
# Preserve extra fields from upstream message (e.g., reasoning_content, refusal, audio, annotations)
original_message = response_json["choices"][0]["message"]
new_message = {
"role": "assistant",
"content": prefix_text,
"tool_calls": tool_calls,
}
# Copy over any extra fields that upstream returned
for key in original_message:
if key not in ["role", "content", "tool_calls"]:
new_message[key] = original_message[key]
response_json["choices"][0]["message"] = new_message
response_json["choices"][0]["finish_reason"] = "tool_calls"
logger.debug(f"🔧 Function call conversion completed")
else:
logger.debug(f"🔧 No tool calls detected, returning original content (including think blocks)")
else:
logger.debug(f"🔧 No function calls detected or conversion conditions not met")
return JSONResponse(content=response_json)
except httpx.HTTPStatusError as e:
logger.error(f"❌ Upstream service response error: status_code={e.response.status_code}")
logger.error(f"❌ Upstream error details: {e.response.text}")
if e.response.status_code == 400:
error_response = {
"error": {
"message": "Invalid request parameters",
"type": "invalid_request_error",
"code": "bad_request"
}
}
elif e.response.status_code == 401:
error_response = {
"error": {
"message": "Authentication failed",
"type": "authentication_error",
"code": "unauthorized"
}
}
elif e.response.status_code == 403:
error_response = {
"error": {
"message": "Access forbidden",
"type": "permission_error",
"code": "forbidden"
}
}
elif e.response.status_code == 429:
error_response = {
"error": {
"message": "Rate limit exceeded",
"type": "rate_limit_error",
"code": "rate_limit_exceeded"
}
}
elif e.response.status_code >= 500:
error_response = {
"error": {
"message": "Upstream service temporarily unavailable",
"type": "service_error",
"code": "upstream_error"
}
}
else:
error_response = {
"error": {
"message": "Request processing failed",
"type": "api_error",
"code": "unknown_error"
}
}
return JSONResponse(content=error_response, status_code=e.response.status_code)
else:
async def stream_with_token_count():
completion_tokens = 0
completion_text = ""
done_received = False
stream_id = None # Keep all streamed chunks under the same id (OpenAI-compatible)
upstream_usage_chunk = None # Store upstream usage chunk if any
async for chunk in stream_proxy_with_fc_transform(
upstream_url,
request_body_dict,
headers,
body.model,
has_function_call,
GLOBAL_TRIGGER_SIGNAL,
request_body_dict["messages"],
tools=body.tools or [],
):
# Check if this is the [DONE] marker
if chunk.startswith(b"data: "):
try:
line_data = chunk[6:].decode('utf-8').strip()
if line_data == "[DONE]":
done_received = True
# Don't yield the [DONE] marker yet, we'll send it after usage info
break
elif line_data:
chunk_json = json.loads(line_data)
if stream_id is None and isinstance(chunk_json, dict):
stream_id = chunk_json.get("id")
if chunk_json.get("object") == "chat.completion.chunk.internal":
raw_fc_content = chunk_json.get("_internal_fc_raw_content", "")
if raw_fc_content:
completion_text += raw_fc_content
logger.debug(f"🔧 Received internal FC raw content for token counting: {len(raw_fc_content)} chars")
continue
# Check if this chunk contains usage information
if "usage" in chunk_json:
upstream_usage_chunk = chunk_json
logger.debug(f"🔧 Detected upstream usage data in chunk")
if not ("choices" in chunk_json and len(chunk_json["choices"]) > 0):
continue
else:
chunk_json = {k: v for k, v in chunk_json.items() if k != "usage"}
chunk = f"data: {json.dumps(chunk_json)}\n\n".encode('utf-8')
# Process regular content chunks
if "choices" in chunk_json and len(chunk_json["choices"]) > 0:
delta = chunk_json["choices"][0].get("delta", {})
# Accumulate content
content = delta.get("content", "")
if content:
completion_text += content
# Accumulate reasoning_content
reasoning_content = delta.get("reasoning_content", "")
if reasoning_content:
completion_text += reasoning_content
logger.debug(f"🔧 Found reasoning_content in stream, accumulating for token count")
except (json.JSONDecodeError, KeyError, UnicodeDecodeError) as e:
logger.debug(f"Failed to parse chunk for token counting: {e}")
pass
yield chunk
# Calculate our estimated tokens
estimated_completion_tokens = token_counter.count_text_tokens(completion_text, body.model) if completion_text else 0
estimated_prompt_tokens = prompt_tokens
estimated_total_tokens = estimated_prompt_tokens + estimated_completion_tokens
elapsed_time = time.time() - start_time
# Determine final usage
final_usage = None
if upstream_usage_chunk and "usage" in upstream_usage_chunk:
# Respect upstream usage, but replace zero values
upstream_usage = upstream_usage_chunk["usage"]
final_usage = upstream_usage.copy()
if not final_usage.get("prompt_tokens") or final_usage.get("prompt_tokens") == 0:
final_usage["prompt_tokens"] = estimated_prompt_tokens
logger.debug(f"🔧 Replaced zero/missing prompt_tokens with estimate: {estimated_prompt_tokens}")
if not final_usage.get("completion_tokens") or final_usage.get("completion_tokens") == 0:
final_usage["completion_tokens"] = estimated_completion_tokens
logger.debug(f"🔧 Replaced zero/missing completion_tokens with estimate: {estimated_completion_tokens}")
if not final_usage.get("total_tokens") or final_usage.get("total_tokens") == 0:
final_usage["total_tokens"] = final_usage.get("prompt_tokens", estimated_prompt_tokens) + final_usage.get("completion_tokens", estimated_completion_tokens)
logger.debug(f"🔧 Replaced zero/missing total_tokens with calculated value: {final_usage['total_tokens']}")
logger.debug(f"🔧 Using upstream usage with replacements: {final_usage}")
else:
# No upstream usage, use our estimates
final_usage = {
"prompt_tokens": estimated_prompt_tokens,
"completion_tokens": estimated_completion_tokens,
"total_tokens": estimated_total_tokens
}
logger.debug(f"🔧 No upstream usage found, using estimates")
# Log token statistics
logger.info("=" * 60)
logger.info(f"📊 Token Usage Statistics - Model: {body.model}")
logger.info(f" Input Tokens: {final_usage['prompt_tokens']}")
logger.info(f" Output Tokens: {final_usage['completion_tokens']}")
logger.info(f" Total Tokens: {final_usage['total_tokens']}")
logger.info(f" Duration: {elapsed_time:.2f}s")
logger.info("=" * 60)
# Send usage information only if requested via stream_options.include_usage
if body.stream_options and body.stream_options.get("include_usage", False):
usage_chunk_to_send = {
"id": (upstream_usage_chunk.get("id") if isinstance(upstream_usage_chunk, dict) else None) or stream_id or f"chatcmpl-{uuid.uuid4().hex}",
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": body.model,
"choices": [],
"usage": final_usage
}
# If upstream provided additional fields in the usage chunk, preserve them
if upstream_usage_chunk:
for key in upstream_usage_chunk:
if key not in ["usage", "choices"] and key not in usage_chunk_to_send:
usage_chunk_to_send[key] = upstream_usage_chunk[key]
yield f"data: {json.dumps(usage_chunk_to_send)}\n\n".encode('utf-8')
logger.debug(f"🔧 Sent usage chunk in stream: {usage_chunk_to_send['usage']}")
# Send [DONE] marker if it was received
if done_received:
yield b"data: [DONE]\n\n"
return StreamingResponse(
stream_with_token_count(),
media_type="text/event-stream"
)
async def _attempt_streaming_fc_retry(
original_content: str,
trigger_signal: str,
messages: List[Dict[str, Any]],
url: str,
headers: Dict[str, str],
model: str,
timeout: int,
tools: Optional[List["Tool"]] = None,
) -> Optional[List[Dict[str, Any]]]:
max_attempts = app_config.features.fc_error_retry_max_attempts
current_content = original_content
current_messages = messages.copy()
def _parse_and_validate(current_content: str) -> tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
parsed = parse_function_calls_xml(current_content, trigger_signal)
if not parsed:
return None, None
validation_error = validate_parsed_tools(parsed, tools or [])
if validation_error:
return None, validation_error
return parsed, None
validation_error: Optional[str] = None
for attempt in range(max_attempts):
# Same rule as non-streaming: avoid retrying if the trigger only appears inside .
if find_last_trigger_signal_outside_think(current_content, trigger_signal) == -1:
logger.debug("🔧 Streaming retry: no trigger signal found outside blocks; aborting retry")
return None
validation_error = None
if attempt == 0:
parsed_tools, validation_error = _parse_and_validate(current_content)
if parsed_tools:
return parsed_tools
if attempt >= max_attempts - 1:
logger.warning(f"⚠️ Streaming FC retry failed after {max_attempts} attempts")
return None
# Classify the failure type to choose the right retry strategy
failure_type = _classify_fc_failure(current_content, trigger_signal)
if failure_type == "no_fc":
return None
error_details = validation_error or _diagnose_fc_parse_error(current_content, trigger_signal)
if failure_type == "truncated":
retry_prompt = get_fc_continuation_prompt(current_content, error_details)
logger.info(f"🔄 Streaming FC output truncated, requesting continuation {attempt + 2}/{max_attempts}")
else:
retry_prompt = get_fc_error_retry_prompt(current_content, error_details)
logger.info(f"🔄 Streaming FC syntax error, requesting rewrite {attempt + 2}/{max_attempts}")
logger.debug(f"🔧 Failure type: {failure_type}, error details: {error_details}")
retry_messages = current_messages + [
{"role": "assistant", "content": current_content},
{"role": "user", "content": retry_prompt}
]
try:
retry_response = await http_client.post(
url,
json={"model": model, "messages": retry_messages, "stream": False},
headers=headers,
timeout=timeout
)
retry_response.raise_for_status()
retry_json = retry_response.json()
if retry_json.get("choices") and len(retry_json["choices"]) > 0:
retry_content = retry_json["choices"][0].get("message", {}).get("content", "")
if failure_type == "truncated" and _is_continuation_response(retry_content, trigger_signal):
current_content = _merge_truncated_and_continuation(current_content, retry_content)
logger.info(f"🔧 Streaming: merged continuation, total length: {len(current_content)}")
else:
current_content = retry_content
current_messages = retry_messages
parsed_tools, validation_error = _parse_and_validate(current_content)
if parsed_tools:
return parsed_tools
else:
logger.warning(f"⚠️ Streaming FC retry response has no valid choices")
return None
except Exception as e:
logger.error(f"❌ Streaming FC retry request failed: {e}")
return None
return None
async def stream_proxy_with_fc_transform(
url: str,
body: dict,
headers: dict,
model: str,
has_fc: bool,
trigger_signal: str,
original_messages: Optional[List[Dict[str, Any]]] = None,
tools: Optional[List["Tool"]] = None,
):
"""
Enhanced streaming proxy, supports dynamic trigger signals, avoids misjudgment within think tags
"""
logger.info(f"📝 Starting streaming response from: {url}")
logger.info(f"📝 Function calling enabled: {has_fc}")
if not has_fc or not trigger_signal:
max_attempts = _get_upstream_retry_attempts()
streamed_any_output = False
for attempt in range(max_attempts):
try:
async with http_client.stream("POST", url, json=body, headers=headers, timeout=app_config.server.timeout) as response:
async for chunk in response.aiter_bytes():
streamed_any_output = True
yield chunk
return
except httpx.RemoteProtocolError:
logger.debug("🔧 Upstream closed connection prematurely, ending stream response")
return
except Exception as exc:
if streamed_any_output or not _is_retriable_upstream_error(exc) or attempt >= max_attempts - 1:
raise
delay = _get_upstream_retry_delay(attempt)
logger.warning(
"⚠️ Upstream stream failed with %s; retrying in %.1fs (%s/%s)",
type(exc).__name__, delay, attempt + 2, max_attempts,
)
await asyncio.sleep(delay)
return
detector = StreamingFunctionCallDetector(trigger_signal)
stream_id = None
pending_finish_reason = None
def _ensure_stream_id(chunk_json: Optional[Dict[str, Any]] = None) -> str:
nonlocal stream_id
if stream_id is None:
upstream_id = chunk_json.get("id") if isinstance(chunk_json, dict) else None
stream_id = upstream_id or f"chatcmpl-passthrough-{uuid.uuid4().hex}"
return stream_id
def _prepare_tool_calls(parsed_tools: List[Dict[str, Any]]):
tool_calls = []
for i, tool in enumerate(parsed_tools):
tool_call_id = f"call_{uuid.uuid4().hex}"
tool_calls.append({
"index": i, "id": tool_call_id, "type": "function",
"function": { "name": tool["name"], "arguments": json.dumps(tool["args"]) }
})
return tool_calls
def _build_tool_call_sse_chunks(parsed_tools: List[Dict[str, Any]], model_id: str, raw_content: str = "") -> List[str]:
tool_calls = _prepare_tool_calls(parsed_tools)
chunks: List[str] = []
if raw_content:
metadata_chunk = {
"object": "chat.completion.chunk.internal",
"_internal_fc_raw_content": raw_content
}
chunks.append(f"data: {json.dumps(metadata_chunk)}\n\n")
tc_id = stream_id or f"chatcmpl-{uuid.uuid4().hex}"
initial_chunk = {
"id": tc_id, "object": "chat.completion.chunk",
"created": int(time.time()), "model": model_id,
"choices": [{"index": 0, "delta": {"role": "assistant", "content": None, "tool_calls": tool_calls}, "finish_reason": None}],
}
chunks.append(f"data: {json.dumps(initial_chunk)}\n\n")
final_chunk = {
"id": tc_id, "object": "chat.completion.chunk",
"created": int(time.time()), "model": model_id,
"choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}],
}
chunks.append(f"data: {json.dumps(final_chunk)}\n\n")
chunks.append("data: [DONE]\n\n")
return chunks
max_attempts = _get_upstream_retry_attempts()
for attempt in range(max_attempts):
try:
async with http_client.stream("POST", url, json=body, headers=headers, timeout=app_config.server.timeout) as response:
if response.status_code != 200:
error_content = await response.aread()
logger.error(f"❌ Upstream service stream response error: status_code={response.status_code}")
logger.error(f"❌ Upstream error details: {error_content.decode('utf-8', errors='ignore')}")
if response.status_code == 401:
error_message = "Authentication failed"
elif response.status_code == 403:
error_message = "Access forbidden"
elif response.status_code == 429:
error_message = "Rate limit exceeded"
elif response.status_code >= 500:
error_message = "Upstream service temporarily unavailable"
else:
error_message = "Request processing failed"
error_chunk = {"error": {"message": error_message, "type": "upstream_error"}}
yield f"data: {json.dumps(error_chunk)}\n\n".encode('utf-8')
yield b"data: [DONE]\n\n"
return
async for line in response.aiter_lines():
if detector.state == "tool_parsing":
if line.startswith("data:"):
line_data = line[len("data: "):].strip()
if line_data and line_data != "[DONE]":
try:
chunk_json = json.loads(line_data)
delta_content = chunk_json.get("choices", [{}])[0].get("delta", {}).get("content", "") or ""
detector.content_buffer += delta_content
# Early termination: once appears, parse and finish immediately
if "" in detector.content_buffer:
logger.debug("🔧 Detected in stream, finalizing early...")
parsed_tools = detector.finalize()
if parsed_tools:
validation_error = validate_parsed_tools(parsed_tools, tools or [])
if validation_error:
logger.info(f"🔧 Tool/schema validation failed in stream finalize: {validation_error}")
parsed_tools = None
if parsed_tools:
logger.debug(f"🔧 Early finalize: parsed {len(parsed_tools)} tool calls")
for sse in _build_tool_call_sse_chunks(parsed_tools, model, detector.content_buffer):
yield sse.encode('utf-8')
return
else:
if app_config.features.enable_fc_error_retry and original_messages:
logger.info(f"🔄 Early finalize FC parsing failed, attempting retry...")
retry_parsed = await _attempt_streaming_fc_retry(
original_content=detector.content_buffer,
trigger_signal=trigger_signal,
messages=original_messages,
url=url,
headers=headers,
model=model,
timeout=app_config.server.timeout,
tools=tools,
)
if retry_parsed:
logger.info(f"✅ Early finalize FC retry succeeded, parsed {len(retry_parsed)} tool calls")
for sse in _build_tool_call_sse_chunks(retry_parsed, model, detector.content_buffer):
yield sse.encode('utf-8')
return
else:
logger.warning(f"⚠️ Early finalize FC retry also failed, ending stream")
else:
logger.warning(
"⚠️ Early finalize detected but failed to parse tool calls; "
"silently ending stream. buffer_len=%s preview=%r",
len(detector.content_buffer),
detector.content_buffer[:200],
)
stop_chunk = {
"id": _ensure_stream_id(),
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]
}
yield f"data: {json.dumps(stop_chunk)}\n\n".encode('utf-8')
yield b"data: [DONE]\n\n"
return
except (json.JSONDecodeError, IndexError):
pass
continue
if line.startswith("data:"):
line_data = line[len("data: "):].strip()
if not line_data or line_data == "[DONE]":
continue
try:
chunk_json = json.loads(line_data)
delta = chunk_json.get("choices", [{}])[0].get("delta", {})
delta_content = delta.get("content", "") or ""
delta_reasoning = delta.get("reasoning_content", "") or ""
finish_reason = chunk_json.get("choices", [{}])[0].get("finish_reason")
# Forward reasoning_content directly (it's not part of function call detection)
if delta_reasoning:
reasoning_chunk = {
"id": _ensure_stream_id(chunk_json),
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {"reasoning_content": delta_reasoning}}]
}
yield f"data: {json.dumps(reasoning_chunk)}\n\n".encode('utf-8')
if delta_content:
is_detected, content_to_yield = detector.process_chunk(delta_content)
if content_to_yield:
yield_chunk = {
"id": _ensure_stream_id(chunk_json),
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {"content": content_to_yield}}]
}
yield f"data: {json.dumps(yield_chunk)}\n\n".encode('utf-8')
if is_detected:
# Tool call signal detected, switch to parsing mode
continue
if finish_reason:
pending_finish_reason = finish_reason
except (json.JSONDecodeError, IndexError):
# Ensure we always yield bytes to keep stream_with_token_count() stable
yield (line + "\n\n").encode("utf-8")
# Stream completed successfully, break out of retry loop
break
except Exception as exc:
if not _is_retriable_upstream_error(exc) or attempt >= max_attempts - 1:
logger.error(f"❌ Failed to connect to upstream service: {exc}")
logger.error(f"❌ Error type: {type(exc).__name__}")
error_message = "Failed to connect to upstream service"
error_chunk = {"error": {"message": error_message, "type": "connection_error"}}
yield f"data: {json.dumps(error_chunk)}\n\n".encode('utf-8')
yield b"data: [DONE]\n\n"
return
delay = _get_upstream_retry_delay(attempt)
logger.warning(
"⚠️ Upstream stream (FC) failed with %s; retrying in %.1fs (%s/%s)",
type(exc).__name__, delay, attempt + 2, max_attempts,
)
await asyncio.sleep(delay)
# Reset detector state for retry
detector = StreamingFunctionCallDetector(trigger_signal)
stream_id = None
pending_finish_reason = None
continue
if detector.state == "tool_parsing":
logger.debug(f"🔧 Stream ended, starting to parse tool call XML...")
parsed_tools = detector.finalize()
if parsed_tools:
validation_error = validate_parsed_tools(parsed_tools, tools or [])
if validation_error:
logger.info(f"🔧 Tool/schema validation failed at stream end: {validation_error}")
parsed_tools = None
if parsed_tools:
logger.debug(f"🔧 Streaming processing: Successfully parsed {len(parsed_tools)} tool calls")
for sse in _build_tool_call_sse_chunks(parsed_tools, model, detector.content_buffer):
yield sse.encode("utf-8")
return
else:
if app_config.features.enable_fc_error_retry and original_messages:
logger.info(f"🔄 Streaming FC parsing failed, attempting retry with error correction...")
retry_parsed = await _attempt_streaming_fc_retry(
original_content=detector.content_buffer,
trigger_signal=trigger_signal,
messages=original_messages,
url=url,
headers=headers,
model=model,
timeout=app_config.server.timeout,
tools=tools,
)
if retry_parsed:
logger.info(f"✅ Streaming FC retry succeeded, parsed {len(retry_parsed)} tool calls")
for sse in _build_tool_call_sse_chunks(retry_parsed, model, detector.content_buffer):
yield sse.encode("utf-8")
return
else:
logger.warning(f"⚠️ Streaming FC retry also failed, falling back to text output")
else:
logger.warning(
"⚠️ Detected tool call signal but XML parsing failed; outputting accumulated text. "
"buffer_len=%s preview=%r",
len(detector.content_buffer),
detector.content_buffer[:300],
)
if detector.content_buffer:
content_chunk = {
"id": _ensure_stream_id(),
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {"content": detector.content_buffer}}]
}
yield f"data: {json.dumps(content_chunk)}\n\n".encode('utf-8')
elif detector.state == "detecting" and detector.content_buffer:
# If stream has ended but buffer still has remaining characters insufficient to form signal, output them
final_yield_chunk = {
"id": _ensure_stream_id(), "object": "chat.completion.chunk",
"created": int(time.time()), "model": model,
"choices": [{"index": 0, "delta": {"content": detector.content_buffer}}]
}
yield f"data: {json.dumps(final_yield_chunk)}\n\n".encode('utf-8')
stop_chunk = {
"id": _ensure_stream_id(),
"object": "chat.completion.chunk",
"created": int(time.time()),
"model": model,
"choices": [{"index": 0, "delta": {}, "finish_reason": pending_finish_reason or "stop"}]
}
yield f"data: {json.dumps(stop_chunk)}\n\n".encode('utf-8')
yield b"data: [DONE]\n\n"
@app.get("/")
def read_root():
return {
"status": "Toolify is running",
"config": {
"upstream_services_count": len(app_config.upstream_services),
"client_keys_count": len(app_config.client_authentication.allowed_keys),
"models_count": len(MODEL_TO_SERVICE_MAPPING),
"features": {
"function_calling": app_config.features.enable_function_calling,
"log_level": app_config.features.log_level,
"convert_developer_to_system": app_config.features.convert_developer_to_system,
"random_trigger": True
}
}
}
@app.get("/v1/models")
async def list_models(_api_key: str = Depends(verify_api_key)):
"""List all available models"""
visible_models = set()
for model_name in MODEL_TO_SERVICE_MAPPING.keys():
if ':' in model_name:
parts = model_name.split(':', 1)
if len(parts) == 2:
alias, _ = parts
visible_models.add(alias)
else:
visible_models.add(model_name)
else:
visible_models.add(model_name)
models = []
for model_id in sorted(visible_models):
models.append({
"id": model_id,
"object": "model",
"created": 1677610602,
"owned_by": "openai",
"permission": [],
"root": model_id,
"parent": None
})
return {
"object": "list",
"data": models
}
def validate_message_structure(messages: List[Dict[str, Any]]) -> bool:
"""Validate if message structure meets requirements"""
try:
valid_roles = ["system", "user", "assistant", "tool"]
if not app_config.features.convert_developer_to_system:
valid_roles.append("developer")
for i, msg in enumerate(messages):
if "role" not in msg:
logger.error(f"❌ Message {i} missing role field")
return False
if msg["role"] not in valid_roles:
logger.error(f"❌ Invalid role value for message {i}: {msg['role']}")
return False
if msg["role"] == "tool":
if "tool_call_id" not in msg:
logger.error(f"❌ Tool message {i} missing tool_call_id field")
return False
content = msg.get("content")
content_info = ""
if content:
if isinstance(content, str):
content_info = f", content=text({len(content)} chars)"
elif isinstance(content, list):
text_parts = [item for item in content if isinstance(item, dict) and item.get('type') == 'text']
image_parts = [item for item in content if isinstance(item, dict) and item.get('type') == 'image_url']
content_info = f", content=multimodal(text={len(text_parts)}, images={len(image_parts)})"
else:
content_info = f", content={type(content).__name__}"
else:
content_info = ", content=empty"
logger.debug(f"✅ Message {i} validation passed: role={msg['role']}{content_info}")
logger.debug(f"✅ All messages validated successfully, total {len(messages)} messages")
return True
except Exception as e:
logger.error(f"❌ Message validation exception: {e}")
return False
def safe_process_tool_choice(tool_choice, tools: Optional[List[Tool]] = None) -> str:
"""
Process tool_choice field and return additional prompt instructions.
Args:
tool_choice: The tool_choice value from the request (str or ToolChoice object)
tools: List of available tools (for validation when specific tool is required)
Returns:
Additional prompt text to append to the function calling prompt
Raises:
HTTPException: If tool_choice specifies a tool that doesn't exist in tools list
"""
try:
if tool_choice is None:
return ""
if isinstance(tool_choice, str):
if tool_choice == "none":
return "\n\n**IMPORTANT:** You are prohibited from using any tools in this round. Please respond like a normal chat assistant and answer the user's question directly."
elif tool_choice == "auto":
# Default behavior, no additional constraints
return ""
elif tool_choice == "required":
return "\n\n**IMPORTANT:** You MUST call at least one tool in this response. Do not respond without using tools."
else:
logger.warning(f"⚠️ Unknown tool_choice string value: {tool_choice}")
return ""
# Handle ToolChoice object: {"type": "function", "function": {"name": "xxx"}}
elif hasattr(tool_choice, 'function'):
function_dict = tool_choice.function
if not isinstance(function_dict, dict):
raise HTTPException(status_code=400, detail="tool_choice.function must be an object")
required_tool_name = function_dict.get("name")
if not required_tool_name or not isinstance(required_tool_name, str):
raise HTTPException(status_code=400, detail="tool_choice.function.name must be a non-empty string")
if not tools:
raise HTTPException(status_code=400, detail="tool_choice requires a non-empty tools list in the request")
tool_names = [t.function.name for t in tools]
if required_tool_name not in tool_names:
raise HTTPException(
status_code=400,
detail=f"tool_choice specifies tool '{required_tool_name}' which is not in the tools list. Available tools: {tool_names}"
)
return f"\n\n**IMPORTANT:** In this round, you must use ONLY the tool named `{required_tool_name}`. Generate the necessary parameters and output in the specified XML format."
else:
logger.warning(f"⚠️ Unsupported tool_choice type: {type(tool_choice)}")
return ""
except HTTPException:
# Re-raise HTTPException to preserve status code
raise
except Exception as e:
logger.error(f"❌ Error processing tool_choice: {e}")
return ""
if __name__ == "__main__":
import uvicorn
logger.info(f"🚀 Starting server on {app_config.server.host}:{app_config.server.port}")
logger.info(f"⏱️ Request timeout: {app_config.server.timeout} seconds")
uvicorn.run(
app,
host=app_config.server.host,
port=app_config.server.port,
log_level=app_config.features.log_level.lower() if app_config.features.log_level != "DISABLED" else "critical"
)