SimpleSallap:SimpleMCP:ToolCalls beyond initial go

Define a typealias for HttpHeaders and use it where ever needed.
Inturn map this to email.message.Message and dict for now.
If and when python evolves Http Headers type into better one,
need to replace in only one place.

Add a ToolManager class which
* maintains the list of tool calls and inturn allows any given
  tool call to be executed and response returned along with needed
  meta data
* generate the overall tool calls meta data
* add ToolCallResponseEx which maintains full TCOutResponse for
  use by tc_handle callers

Avoid duplicating handling of some of the basic needed http header
entries.

Move checking for any dependencies before enabling a tool call into
respective tc??? module.
* for now this also demotes the logic from the previous fine grained
  per tool call based dependency check to a more global dep check at
  the respective module level
This commit is contained in:
hanishkvc 2025-12-06 21:50:51 +05:30
parent 0a445c875b
commit 8700d522a5
4 changed files with 64 additions and 36 deletions

View File

@ -9,7 +9,7 @@ from dataclasses import dataclass
def get_from_web(url: str, tag: str, inContentType: str, inHeaders: dict[str, str|None]):
def get_from_web(url: str, tag: str, inContentType: str, inHeaders: mTC.HttpHeaders):
"""
Get the url specified from web.
@ -58,7 +58,7 @@ def get_from_local(urlParts: urllib.parse.ParseResult, tag: str, inContentType:
return mTC.TCOutResponse(False, 502, f"WARN:{tag}:Failed:{exc}")
def get_file(url: str, tag: str, inContentType: str, inHeaders: dict[str, str|None]={}):
def get_file(url: str, tag: str, inContentType: str, inHeaders: mTC.HttpHeaders={}):
"""
Based on the scheme specified in the passed url,
either get from local file system or from the web.

View File

@ -1,12 +1,10 @@
# Helper to manage pdf related requests
# by Humans for All
import urllib.parse
import urlvalidator as uv
import filemagic as mFile
import toolcall as mTC
import http.client
from typing import TYPE_CHECKING, Any
from typing import Any
PDFOUTLINE_MAXDEPTH=4
@ -92,7 +90,7 @@ class TCPdfText(mTC.ToolCall):
)
)
def tc_handle(self, args: mTC.TCInArgs, inHeaders: http.client.HTTPMessage) -> mTC.TCOutResponse:
def tc_handle(self, args: mTC.TCInArgs, inHeaders: mTC.HttpHeaders) -> mTC.TCOutResponse:
"""
Handle pdftext request,
which is used to extract plain text from the specified pdf file.
@ -105,3 +103,14 @@ class TCPdfText(mTC.ToolCall):
return process_pdftext(url, startP, endP)
except Exception as exc:
return mTC.TCOutResponse(False, 502, f"WARN:HandlePdfText:Failed:{exc}")
def ok():
import importlib
dep = "pypdf"
try:
importlib.import_module(dep)
return True
except ImportError as exc:
print(f"WARN:TCPdf:{dep} missing or has issues, so not enabling myself")
return False

View File

@ -1,20 +1,18 @@
# Helper to manage web related requests
# by Humans for All
import urllib.parse
import urlvalidator as uv
import html.parser
import debug
import filemagic as mFile
import json
import re
import http.client
from typing import Any, cast
import toolcall as mTC
def handle_urlreq(url: str, inHeaders: http.client.HTTPMessage, tag: str):
def handle_urlreq(url: str, inHeaders: mTC.HttpHeaders, tag: str):
"""
Common part of the url request handling used by both urlraw and urltext.
@ -33,16 +31,8 @@ def handle_urlreq(url: str, inHeaders: http.client.HTTPMessage, tag: str):
if not gotVU.callOk:
return mTC.TCOutResponse(gotVU.callOk, gotVU.statusCode, gotVU.statusMsg)
try:
hUA = inHeaders.get('User-Agent', None)
hAL = inHeaders.get('Accept-Language', None)
hA = inHeaders.get('Accept', None)
headers = {
'User-Agent': hUA,
'Accept': hA,
'Accept-Language': hAL
}
# Get requested url
return mFile.get_file(url, tag, "text/html", headers)
return mFile.get_file(url, tag, "text/html", inHeaders)
except Exception as exc:
return mTC.TCOutResponse(False, 502, f"WARN:{tag}:Failed:{exc}")
@ -65,7 +55,7 @@ class TCUrlRaw(mTC.ToolCall):
)
)
def tc_handle(self, args: mTC.TCInArgs, inHeaders: http.client.HTTPMessage) -> mTC.TCOutResponse:
def tc_handle(self, args: mTC.TCInArgs, inHeaders: mTC.HttpHeaders) -> mTC.TCOutResponse:
try:
# Get requested url
got = handle_urlreq(args['url'], inHeaders, "HandleTCUrlRaw")
@ -212,7 +202,7 @@ class TCHtmlText(mTC.ToolCall):
)
)
def tc_handle(self, args: mTC.TCInArgs, inHeaders: http.client.HTTPMessage) -> mTC.TCOutResponse:
def tc_handle(self, args: mTC.TCInArgs, inHeaders: mTC.HttpHeaders) -> mTC.TCOutResponse:
try:
# Get requested url
got = handle_urlreq(args['url'], inHeaders, "HandleTCHtmlText")
@ -329,7 +319,7 @@ class TCXmlFiltered(mTC.ToolCall):
)
)
def tc_handle(self, args: mTC.TCInArgs, inHeaders: http.client.HTTPMessage) -> mTC.TCOutResponse:
def tc_handle(self, args: mTC.TCInArgs, inHeaders: mTC.HttpHeaders) -> mTC.TCOutResponse:
try:
# Get requested url
got = handle_urlreq(args['url'], inHeaders, "HandleTCXMLFiltered")
@ -347,3 +337,7 @@ class TCXmlFiltered(mTC.ToolCall):
return mTC.TCOutResponse(True, got.statusCode, got.statusMsg, got.contentType, xmlFiltered.text.encode('utf-8'))
except Exception as exc:
return mTC.TCOutResponse(False, 502, f"WARN:XMLFiltered:Failed:{exc}")
def ok():
return True

View File

@ -1,11 +1,11 @@
# Tool Call Base
# by Humans for All
from typing import Any, TypeAlias
from typing import Any, TypeAlias, TYPE_CHECKING
from dataclasses import dataclass
import http
import http.client
import urllib.parse
if TYPE_CHECKING:
import email.message
#
@ -61,13 +61,6 @@ class ToolCallMeta():
type: str = "function"
function: TCFunction|None = None
@dataclass
class TollCallResponse():
status: bool
tcid: str
name: str
content: str = ""
@dataclass(frozen=True)
class TCOutResponse:
"""
@ -79,6 +72,21 @@ class TCOutResponse:
contentType: str = ""
contentData: bytes = b""
@dataclass
class ToolCallResponseEx():
tcid: str
name: str
response: TCOutResponse
@dataclass
class ToolCallResponse():
status: bool
tcid: str
name: str
content: str = ""
HttpHeaders: TypeAlias = dict[str, str] | email.message.Message[str, str]
@dataclass
class ToolCall():
@ -87,13 +95,30 @@ class ToolCall():
def tcf_meta(self) -> TCFunction|None:
return None
def tc_handle(self, args: TCInArgs, inHeaders: http.client.HTTPMessage) -> TCOutResponse:
def tc_handle(self, args: TCInArgs, inHeaders: HttpHeaders) -> TCOutResponse:
return TCOutResponse(False, 500)
def meta(self) -> ToolCallMeta:
tcf = self.tcf_meta()
return ToolCallMeta("function", tcf)
def handler(self, callId: str, args: Any, inHeaders: http.client.HTTPMessage) -> TollCallResponse:
got = self.tc_handle(args, inHeaders)
return TollCallResponse(got.callOk, callId, self.name, got.contentData.decode('utf-8'))
class ToolManager():
def __init__(self) -> None:
self.toolcalls: dict[str, ToolCall] = {}
def tc_add(self, fName: str, tc: ToolCall):
self.toolcalls[fName] = tc
def meta(self):
oMeta = {}
for tcName in self.toolcalls.keys():
oMeta[tcName] = self.toolcalls[tcName].meta()
def tc_handle(self, tcName: str, callId: str, tcArgs: TCInArgs, inHeaders: HttpHeaders) -> ToolCallResponseEx:
try:
response = self.toolcalls[tcName].tc_handle(tcArgs, inHeaders)
return ToolCallResponseEx(callId, tcName, response)
except KeyError:
return ToolCallResponseEx(callId, tcName, TCOutResponse(False, 400, "Unknown tool call"))