import html
import logging
import re
import unicodedata
from typing import Any
from urllib.parse import quote, unquote, urlsplit
import httpx
from fastapi import FastAPI
from models import AuthorWorksInput, ChatToolResponse, GetWorkInput, SearchWorksInput
CROSSREF_BASE = "https://api.crossref.org"
TIMEOUT = 20.0
logger = logging.getLogger(__name__)
_REQUEST_FAILED_MESSAGE = "Crossref request failed. Please try again later."
app = FastAPI(
title="Crossref Omi Integration",
description="No-auth Crossref chat tools for paper metadata search and lookup",
version="1.0.1",
)
def clamp_max_results(value: int) -> int:
return max(1, min(10, value))
_DOI_PREFIX_RE = re.compile(r"^10\.[0-9]{4,9}(?:\.[0-9]+)*/")
def _is_valid_doi(value: str) -> bool:
"""Accept visible DOI characters; reject whitespace and non-graphic code points."""
prefix = _DOI_PREFIX_RE.match(value)
if prefix is None:
return False
suffix = value[prefix.end() :]
return bool(suffix) and all(unicodedata.category(character)[0] in "LMNPS" for character in suffix)
def normalize_doi(value: Any) -> str | None:
"""Normalize a DOI identifier before using it as a Crossref path segment.
Chat callers commonly paste resolver links instead of the bare DOI. Only
the two DOI resolver hosts are accepted; their query and fragment are
discarded before decoding the path exactly once. Any remaining percent
escape is kept literal and safely re-encoded for the Crossref API request.
"""
if not isinstance(value, str):
return None
raw = value.strip()
if not raw:
return None
if raw.lower().startswith("doi:"):
raw = raw[4:].strip()
elif "://" in raw or raw.lower().startswith(("http:", "https:")):
try:
parsed = urlsplit(raw)
host = (parsed.hostname or "").lower().rstrip(".")
if parsed.scheme.lower() not in {"http", "https"}:
return None
if host not in {"doi.org", "dx.doi.org"}:
return None
if parsed.username or parsed.password:
return None
if parsed.port not in (None, 80, 443):
return None
except ValueError:
return None
raw = unquote(parsed.path.lstrip("/"))
if not _is_valid_doi(raw):
return None
return raw
_JATS_TAG = re.compile(r"</?jats:[^>]+>")
_FACE_TAG_PAIR = re.compile(
r"<(?P<tag>b|i|u|sub|sup|scp|tt|font|sc|strike)(?:\s+[^>]*)?>(.*?)</(?P=tag)>",
re.IGNORECASE | re.DOTALL,
)
_CLOSE_TAG = re.compile(r"</[a-zA-Z][^>]*>")
_OPEN_TAG = re.compile(r"(?<![A-Za-z0-9_])<[a-zA-Z][^>]*>")
def clean(text: Any) -> str:
if text is None:
return ""
value = html.unescape(str(text)).strip()
value = _JATS_TAG.sub("", value)
prev = None
while prev != value:
prev = value
value = _FACE_TAG_PAIR.sub(r"\2", value)
value = _CLOSE_TAG.sub("", value)
value = _OPEN_TAG.sub("", value)
return value.strip()
def extract_year(item: dict[str, Any]) -> str:
for key in ("published-print", "published-online", "issued"):
date_parts = (item.get(key) or {}).get("date-parts", [])
if date_parts and date_parts[0]:
return clean(date_parts[0][0])
return ""
async def crossref_get(path: str, params: dict[str, Any]) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
response = await client.get(f"{CROSSREF_BASE}{path}", params=params)
response.raise_for_status()
return response.json()
@app.get("/health")
async def health():
return {"status": "ok"}
@app.get("/tools")
async def tools():
return {
"tools": [
{
"name": "search_crossref_works",
"description": "Search scholarly works by keyword via Crossref",
"parameters": {
"query": {"type": "string", "description": "Search keyword(s)"},
"max_results": {
"type": "integer",
"description": "Number of results (1-10)",
"default": 5,
},
},
},
{
"name": "get_crossref_work",
"description": "Get details for a specific work by DOI",
"parameters": {
"doi": {
"type": "string",
"description": "DOI, e.g. 10.1038/nphys1170",
}
},
},
{
"name": "get_crossref_works_by_author",
"description": "Find recent works for an author name",
"parameters": {
"author": {"type": "string", "description": "Author name"},
"max_results": {
"type": "integer",
"description": "Number of results (1-10)",
"default": 5,
},
},
},
]
}
@app.get("/.well-known/omi-tools.json")
async def get_omi_tools_manifest():
return {
"tools": [
{
"name": "search_crossref_works",
"description": "Search scholarly works by keyword via Crossref",
"endpoint": "/tools/search_crossref_works",
"method": "POST",
"parameters": {
"properties": {
"query": {"type": "string", "description": "Search keyword(s)"},
"max_results": {
"type": "integer",
"description": "Number of results (1-10)",
"default": 5,
},
},
"required": ["query"],
},
"auth_required": False,
"status_message": "Searching Crossref...",
},
{
"name": "get_crossref_work",
"description": "Get details for a specific work by DOI",
"endpoint": "/tools/get_crossref_work",
"method": "POST",
"parameters": {
"properties": {
"doi": {
"type": "string",
"description": "DOI, e.g. 10.1038/nphys1170",
}
},
"required": ["doi"],
},
"auth_required": False,
"status_message": "Fetching Crossref work...",
},
{
"name": "get_crossref_works_by_author",
"description": "Find recent works for an author name",
"endpoint": "/tools/get_crossref_works_by_author",
"method": "POST",
"parameters": {
"properties": {
"author": {"type": "string", "description": "Author name"},
"max_results": {
"type": "integer",
"description": "Number of results (1-10)",
"default": 5,
},
},
"required": ["author"],
},
"auth_required": False,
"status_message": "Fetching author works...",
},
]
}
@app.post("/tools/search_crossref_works", response_model=ChatToolResponse)
async def search_crossref_works(payload: SearchWorksInput):
query = payload.query.strip()
if len(query) < 2:
return ChatToolResponse(error="Query must be at least 2 characters.")
limited = clamp_max_results(payload.max_results)
try:
payload = await crossref_get(
"/works",
{"query": query, "rows": limited, "sort": "relevance", "order": "desc"},
)
except Exception as exc:
logger.warning("Crossref search request failed: %s", type(exc).__name__)
return ChatToolResponse(error=_REQUEST_FAILED_MESSAGE)
items = payload.get("message", {}).get("items", [])
if not items:
return ChatToolResponse(result=f"No Crossref results found for '{query}'.")
lines = [f"Top {len(items)} Crossref results for '{query}':"]
for idx, item in enumerate(items, 1):
title = clean((item.get("title") or ["Untitled"])[0])
doi = clean(item.get("DOI"))
year = extract_year(item)
lines.append(f"{idx}. {title} ({year})")
lines.append(f" DOI: {doi}")
return ChatToolResponse(result="\n".join(lines))
@app.post("/tools/get_crossref_work", response_model=ChatToolResponse)
async def get_crossref_work(payload: GetWorkInput):
normalized = normalize_doi(payload.doi)
if normalized is None:
return ChatToolResponse(error="Invalid DOI format. Example: 10.1038/nphys1170")
try:
payload = await crossref_get(f"/works/{quote(normalized, safe='')}", {})
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 404:
return ChatToolResponse(error=f"No Crossref record found for DOI '{normalized}'.")
logger.warning("Crossref work request failed with status %s", exc.response.status_code)
return ChatToolResponse(error=_REQUEST_FAILED_MESSAGE)
except Exception as exc:
logger.warning("Crossref work request failed: %s", type(exc).__name__)
return ChatToolResponse(error=_REQUEST_FAILED_MESSAGE)
item = payload.get("message", {})
title = clean((item.get("title") or ["Untitled"])[0])
publisher = clean(item.get("publisher"))
doi_out = clean(item.get("DOI"))
url = clean(item.get("URL"))
abstract = clean(item.get("abstract"))
year = extract_year(item)
parts = [
f"Title: {title}",
f"DOI: {doi_out}",
f"Year: {year}",
f"Publisher: {publisher}",
f"URL: {url}",
]
if abstract:
parts.append(f"Abstract: {abstract[:1200]}")
return ChatToolResponse(result="\n".join(parts))
@app.post("/tools/get_crossref_works_by_author", response_model=ChatToolResponse)
async def get_crossref_works_by_author(payload: AuthorWorksInput):
author = payload.author.strip()
if len(author) < 2:
return ChatToolResponse(error="Author must be at least 2 characters.")
limited = clamp_max_results(payload.max_results)
try:
payload = await crossref_get(
"/works",
{
"query.author": author,
"rows": limited,
"sort": "published",
"order": "desc",
},
)
except Exception as exc:
logger.warning("Crossref author request failed: %s", type(exc).__name__)
return ChatToolResponse(error=_REQUEST_FAILED_MESSAGE)
items = payload.get("message", {}).get("items", [])
if not items:
return ChatToolResponse(result=f"No recent works found for author '{author}'.")
lines = [f"Recent works for '{author}':"]
for idx, item in enumerate(items, 1):
title = clean((item.get("title") or ["Untitled"])[0])
doi = clean(item.get("DOI"))
year = extract_year(item)
lines.append(f"{idx}. {title} ({year})")
lines.append(f" DOI: {doi}")
return ChatToolResponse(result="\n".join(lines))